lib/redis/stream/client.rb
def sync_add(data = {}, options = {})
        raise "Client isn't running" unless @state.eql?(Redis::Stream::State::RUNNING)

        default_options = {"to" => "*", "group" => "*", "type" => Redis::Stream::Type::ACTION, "time_out" => 5, "passthrough" => false, "cache_key" => nil, "tracer" => nil}
        options = default_options.merge(options)

        to = options["to"]
        group = options["group"]
        passthrough = options["passthrough"]
        time_out = options["time_out"]

        #@state = Redis::Stream::State::RUNNING
        data_out = nil
        add_id = add(data, "to" => to, "group" => group, "type" => options["type"], "cache_key" => options["cache_key"], "tracer" => options['tracer'])

        time = Time.now

        loop do
          timing = ((Time.now - time)).to_i
          if timing > time_out
            @logger.info("#{@consumer_id} - Time out(#{time_out}) for '#{to}' in group '#{group}'")
            #@send_queue.delete(add_id) if @send_queue.include?(add_id)
            break
          end
          break if (data_out = read_next_message_from_stream(false, passthrough))
        end
        #@state = Redis::Stream::State::STOPPED
        data_out
      end
# sync_add: same as add command but synchronous. Blocks call until a message arrives
# @param [Object] data                    Any data you want to transmit
# @param [Hash{String->String}]           Options
same as add command but synchronous. Blocks call until a message arrives
same as add command but synchronous . Blocks call until a message arrives
def sync_add(data = {}, options = {})
        raise "Client isn't running" unless @state.eql?(Redis::Stream::State::RUNNING)
        default_options = {"to" => "*", "group" => "*", "type" => Redis::Stream::Type::ACTION, "time_out" => 5, "passthrough" => false, "cache_key" => nil, "tracer" => nil}
        options = default_options.merge(options)
        to = options["to"]
        group = options["group"]
        passthrough = options["passthrough"]
        time_out = options["time_out"]
        data_out = nil
        add_id = add(data, "to" => to, "group" => group, "type" => options["type"], "cache_key" => options["cache_key"], "tracer" => options['tracer'])
        time = Time.now
        loop do
          timing = ((Time.now - time)).to_i
          if timing > time_out
            @logger.info("#{@consumer_id} - Time out(#{time_out}) for '#{to}' in group '#{group}'")
            break
          end
          break if (data_out = read_next_message_from_stream(false, passthrough))
        end
        data_out
      end
def sync_add ( data = { } , options = { } ) raise "Client isn't running" unless @state . eql? ( Redis :: Stream :: State :: RUNNING ) default_options = { "to" => "*" , "group" => "*" , "type" => Redis :: Stream :: Type :: ACTION , "time_out" => 5 , "passthrough" => false , "cache_key" => nil , "tracer" => nil } options = default_options . merge ( options ) to = options [ "to" ] group = options [ "group" ] passthrough = options [ "passthrough" ] time_out = options [ "time_out" ] data_out = nil add_id = add ( data , "to" => to , "group" => group , "type" => options [ "type" ] , "cache_key" => options [ "cache_key" ] , "tracer" => options [ 'tracer' ] ) time = Time . now loop do timing = ( ( Time . now - time ) ) . to_i if timing > time_out @logger . info ( "#{@consumer_id} - Time out(#{time_out}) for '#{to}' in group '#{group}'" ) break end break if ( data_out = read_next_message_from_stream ( false , passthrough ) ) end data_out end
sync_add: same as add command but synchronous.
sync_add : same as add command but synchronous .
#@state = Redis::Stream::State::RUNNING #@send_queue.delete(add_id) if @send_queue.include?(add_id) #@state = Redis::Stream::State::STOPPED
data
options
None