lib/firehose/server/channel_subscription.rb
# Can be raised if a Subscription failed for any reason.
Can be raised if a Subscription failed for any reason.
Can be raised if a Subscription failed for any reason .
def next_messages
        list_key     = Server::Redis.key(channel_key, :list)
        sequence_key = Server::Redis.key(channel_key, :sequence)
        @redis.multi
          @redis.get(sequence_key).
            errback {|e| @deferrable.fail e }
          @redis.lrange(list_key, 0, -1).
            errback {|e| @deferrable.fail e }
        @redis.exec.callback do |(channel_sequence, message_list)|
          message_list.reverse!
          channel_sequence = [channel_sequence.to_i, message_list.size].max
          buffer = MessageBuffer.new(message_list, channel_sequence, @sequence)
          if buffer.remaining_messages.empty?
            Firehose.logger.debug "No messages in buffer, subscribing. sequence: `#{channel_sequence}` consumer_sequence: #{@sequence}"
            subscribe
          else 
            @deferrable.succeed process_messages(buffer.remaining_messages)
            @deferrable.callback { on_unsubscribe }
          end
        end.errback {|e| @deferrable.fail e }
        @deferrable
      end
def next_messages list_key = Server :: Redis . key ( channel_key , :list ) sequence_key = Server :: Redis . key ( channel_key , :sequence ) @redis . multi @redis . get ( sequence_key ) . errback { | e | @deferrable . fail e } @redis . lrange ( list_key , 0 , - 1 ) . errback { | e | @deferrable . fail e } @redis . exec . callback do | ( channel_sequence , message_list ) | message_list . reverse! channel_sequence = [ channel_sequence . to_i , message_list . size ] . max buffer = MessageBuffer . new ( message_list , channel_sequence , @sequence ) if buffer . remaining_messages . empty? Firehose . logger . debug "No messages in buffer, subscribing. sequence: `#{channel_sequence}` consumer_sequence: #{@sequence}" subscribe else @deferrable . succeed process_messages ( buffer . remaining_messages ) @deferrable . callback { on_unsubscribe } end end . errback { | e | @deferrable . fail e } @deferrable end
Can be raised if a Subscription failed for any reason.
Can be raised if a Subscription failed for any reason .
# Fetch entire list: http://stackoverflow.com/questions/10703019/redis-fetch-all-value-of-list-without-iteration-and-without-popping # Reverse the messages so they can be correctly procesed by the MessageBuffer class. There's # a patch in the message-buffer-redis branch that moves this concern into the Publisher LUA # script. We kept it out of this for now because it represents a deployment risk and `reverse!` # is a cheap operation in Ruby. # Either this resource has never been seen before or we are all caught up. # Subscribe and hope something gets published to this end-point. # Either the client is under water or caught up to head. # TODO -ARRRG!!!! We can't listen to deferrables here. Need to expose those # on this class because this is a class var. In practice thats how we roll anyway.