lib/isono/node_modules/rpc_channel.rb
def register_endpoint(endpoint, app, opts={})
        raise TypeError unless app.respond_to?(:call)
        opts = {:exclusive=>true, :prefetch=>1}.merge(opts)

        # create receive queue for new RPC endpoint.
        endpoint_proc = proc { |header, data|

          data = Serializer.instance.unmarshal(data)

          resctx = if data[:oneshot]
                     OneshotResponseContext.new(@endpoints[endpoint][:ch].response_exchange, header)
                   else
                     ResponseContext.new(@endpoints[endpoint][:ch].response_exchange, header)
                   end
          begin
            req = Rack::Request.new({:sender=>header.reply_to['command-recv.'.size..-1],
                                      :message_id=>header.message_id
                                    }.merge(data))
            res = Rack::Response.new(resctx)
            ret = app.call(req, res)
          rescue ::Exception => e
            logger.error(e)
            resctx.response(e) unless resctx.responded?
          end
        }


        EventMachine.schedule {
          ch = node.create_channel

          ch.instance_eval %Q{
            def endpoint_queue
              self.queue("isono.rpc.endpoint.#{endpoint}", {:exclusive=>false, :auto_delete=>true})
            end

            def response_exchange
              self.default_exchange
            end
          }
          ch.prefetch(opts[:prefetch].to_i) if opts[:prefetch].to_i > 0
          # stores hash here that is for thread safety.
          if !@endpoints.nil?
            @endpoints[endpoint]={:app=>app, :opts=>opts, :ch=>ch}

            ch.endpoint_queue.subscribe(:ack=>true, &endpoint_proc)
            event.publish('rpc/register', :args=>[endpoint])
          else
            logger.error("No such endpoints")
          end
        }
      end
# Register a new RPC endpoint.
#
# This method works in sync mode if called at non-EM reactor thread.
# @param [String] endpoint
# @param [has call() method]  app
# @param [Hash] opts
#    :exclusive
#    :prefetch
Register a new RPC endpoint.
This method works in sync mode if called at non-EM reactor thread.
Register a new RPC endpoint . This method works in sync mode if called at non - EM reactor thread .
def register_endpoint(endpoint, app, opts={})
        raise TypeError unless app.respond_to?(:call)
        opts = {:exclusive=>true, :prefetch=>1}.merge(opts)
        endpoint_proc = proc { |header, data|
          data = Serializer.instance.unmarshal(data)
          resctx = if data[:oneshot]
                     OneshotResponseContext.new(@endpoints[endpoint][:ch].response_exchange, header)
                   else
                     ResponseContext.new(@endpoints[endpoint][:ch].response_exchange, header)
                   end
          begin
            req = Rack::Request.new({:sender=>header.reply_to['command-recv.'.size..-1],
                                      :message_id=>header.message_id
                                    }.merge(data))
            res = Rack::Response.new(resctx)
            ret = app.call(req, res)
          rescue ::Exception => e
            logger.error(e)
            resctx.response(e) unless resctx.responded?
          end
        }
        EventMachine.schedule {
          ch = node.create_channel
          ch.instance_eval %Q{
            def endpoint_queue
              self.queue("isono.rpc.endpoint.#{endpoint}", {:exclusive=>false, :auto_delete=>true})
            end
            def response_exchange
              self.default_exchange
            end
          }
          ch.prefetch(opts[:prefetch].to_i) if opts[:prefetch].to_i > 0
          if !@endpoints.nil?
            @endpoints[endpoint]={:app=>app, :opts=>opts, :ch=>ch}
            ch.endpoint_queue.subscribe(:ack=>true, &endpoint_proc)
            event.publish('rpc/register', :args=>[endpoint])
          else
            logger.error("No such endpoints")
          end
        }
      end
def register_endpoint ( endpoint , app , opts = { } ) raise TypeError unless app . respond_to? ( :call ) opts = { :exclusive => true , :prefetch => 1 } . merge ( opts ) endpoint_proc = proc { | header , data | data = Serializer . instance . unmarshal ( data ) resctx = if data [ :oneshot ] OneshotResponseContext . new ( @endpoints [ endpoint ] [ :ch ] . response_exchange , header ) else ResponseContext . new ( @endpoints [ endpoint ] [ :ch ] . response_exchange , header ) end begin req = Rack :: Request . new ( { :sender => header . reply_to [ 'command-recv.' . size .. - 1 ] , :message_id => header . message_id } . merge ( data ) ) res = Rack :: Response . new ( resctx ) ret = app . call ( req , res ) rescue :: Exception => e logger . error ( e ) resctx . response ( e ) unless resctx . responded? end } EventMachine . schedule { ch = node . create_channel ch . instance_eval %Q{
            def endpoint_queue
              self.queue("isono.rpc.endpoint.#{endpoint}", {:exclusive=>false, :auto_delete=>true})
            end

            def response_exchange
              self.default_exchange
            end
          } ch . prefetch ( opts [ :prefetch ] . to_i ) if opts [ :prefetch ] . to_i > 0 if ! @endpoints . nil? @endpoints [ endpoint ] = { :app => app , :opts => opts , :ch => ch } ch . endpoint_queue . subscribe ( :ack => true , & endpoint_proc ) event . publish ( 'rpc/register' , :args => [ endpoint ] ) else logger . error ( "No such endpoints" ) end } end
Register a new RPC endpoint.
Register a new RPC endpoint .
# create receive queue for new RPC endpoint. # stores hash here that is for thread safety.
endpoint
app
opts
opts :exclusive :prefetch
opts : exclusive : prefetch