Class: Goat::ChannelPusher

Inherits:
Object show all
Includes:
EM::Deferrable
Defined in:
lib/goat.rb

Instance Method Summary collapse

Constructor Details

#initialize(channel, jsonp) ⇒ ChannelPusher

Returns a new instance of ChannelPusher.



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/goat.rb', line 184

def initialize(channel, jsonp)
  @channel = channel
  @finished = false
  @jsonp = jsonp
  @messages = []

  self.errback do
    Goat.logd "Channel closed"
    @finished = true
  end

  self.callback do
    @finished = true
    @channel.emchannel.unsubscribe(@subscription) if @subscription
  end

  @subscription = @channel.emchannel.subscribe do |msg|
    @messages << msg
    # send on next tick so we get a batch of messages together, rather than a single message
    EM.next_tick { self.send_and_finish }
  end
end

Instance Method Details

#each(&blk) ⇒ Object 



230
231
232
 
# File 'lib/goat.rb', line 230

def each(&blk)
  @body_callback = blk
end

#messages_without_duplicates ⇒ Object 



207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/goat.rb', line 207

def messages_without_duplicates
  @ids = Set.new
  @messages.select do |x|
    if x.include?(:id)
      id = x[:id]
      if @ids.include?(id)
        nil
      else
        @ids << id
        x
      end
    else
      x
    end
  end.compact
end

#send_and_finish ⇒ Object 



224
225
226
227
228
 
# File 'lib/goat.rb', line 224

def send_and_finish
  return if @finished # may be called several times
  @body_callback.call("#{@jsonp}(#{{'messages' => messages_without_duplicates}.to_json})")
  succeed
end