在节俭处理程序函数中获取对等地址

Posted 2023-02-16

【中文标题】在节俭处理程序函数中获取对等地址

【英文标题】：Get peer address within a thrift handler function

【发布时间】：2011-10-02 02:14:32

【问题描述】：

我在 ruby​​ 中实现了一个小型 thrift (0.6.0) 服务器，以充当另一个协议的代理角色，该协议具有到单个服务器的多个连接（多个客户端）。我希望能够在服务器端保留每个客户端的数据，并跨处理函数的多次调用跟踪“会话”参数。

我目前使用Thrift::NonblockingServer，因为SimpleServer 似乎不允许并发连接。

我知道如何使用TCPSocket::peeraddr，但Thrift::NonblockingServer::IOManager::Worker::run 使用它读取的帧创建了一个临时MemoryBufferTransport，并将其作为输入/输出协议传递给处理器，因此信息似乎不是从在那里。

有没有一种干净的方法可以做到这一点？ 我正在考虑重新定义上面提到的Thrift::NonblockingServer::IOManager::Worker::run 以在附加参数中包含 fd 或其他 ID 以处理或增加 proto 实例，但因为我还必须担心一层生成的 ruby​​ 代码（@987654328 @method in class Processor) 好像有点重。

我想知道以前是否有人做过这样的事情。

谢谢！

附言这与this C++ thrift question 的问题类似

【参考方案1】：

以下是我如何更改 Thrift::NonblockingServer::IOManager::Worker::run 以支持此功能。

几个注意事项：

正如我在问题中提到的那样，我认为它并不干净（如果没有其他问题，我将不得不监视未来的节俭版本以了解此功能的变化，这是基于 0.6.0）。 这是为 ruby​​ 1.8 编写的（否则我会得到未翻译的地址/端口 see my other question） 我是 ruby​​ 新手 .. 我确定我做错了一些“错误”（例如，$connections 应该在 ConnEntry 或 diff 类中成为 @@connections 吗？）。李> 我知道线程本地存储的Thread.current 哈希有a namespace pollution issue

首先，在某个中心模块：

module MyThriftExt

  $connections=

  class ConnEntry
    attr_reader :addr_info, :attr

    def initialize(thrift_fd)
      @addr_info=thrift_fd.handle.peeraddr
      @attr=
    end

    def self.PreHandle(fd)
      $connections[fd]=ConnEntry.new(fd) unless $connections[fd].is_a? ConnEntry
      # make the connection entry as short-term thread-local variable 
      # (cleared in postHandle)
      Thread.current[:connEntry]=$connections[fd]
    end

    def self.PostHandle()
      Thread.current[:connEntry]=nil
    end

    def to_s()
      "#addr_info"
    end   end end


module Thrift   class NonblockingServer
    class IOManager
      alias :old_remove_connection :remove_connection
      def remove_connection(fd)
        $connections.delete fd
        old_remove_connection(fd)
      end

      class Worker
        # The following is verbatim from thrift 0.6.0 except for the two lines 
        # marked with "Added"
        def run
          loop do
            cmd, *args = @queue.pop
            case cmd
            when :shutdown
              @logger.debug "#self is shutting down, goodbye"
              break
            when :frame
              fd, frame = args
              begin
                otrans = @transport_factory.get_transport(fd)
                oprot = @protocol_factory.get_protocol(otrans)
                membuf = MemoryBufferTransport.new(frame)
                itrans = @transport_factory.get_transport(membuf)
                iprot = @protocol_factory.get_protocol(itrans)
                MyThriftExt::ConnEntry.PreHandle(fd)    # <<== Added
                @processor.process(iprot, oprot)
                MyThriftExt::ConnEntry.PostHandle       # <<== Added
              rescue => e
                @logger.error "#Thread.current.inspect raised error: #e.inspect\n#e.backtrace.join("\n")"
              end
            end
          end
        end
      end
    end   
  end
end

然后，您可以在处理程序中的任何位置访问 Thread.current[:connEntry].addr_info 以获取特定于连接的数据，或将任何与连接相关的内容存储在 Thread.current[:connEntry].attr 哈希中。