Class: DataSift::LiveStream

Inherits:
ApiResource show all
Defined in:
lib/live_stream.rb

Constant Summary

Constants inherited from ApiResource

ApiResource::TLSv1, ApiResource::TLSv1_2

Constants included from DataSift

APPLICATION_JSON, DELETE, DETECT_DEAD_SOCKETS, GET, HEAD, IS_WINDOWS, KNOWN_SOCKETS, SOCKET_DETECTOR_TIMEOUT, VERSION, X_ANALYSIS_TASKS_QUEUED, X_ANALYSIS_TASKS_QUEUE_LIMIT, X_INSIGHT_TASKS_QUEUED, X_INSIGHT_TASKS_QUEUE_LIMIT, X_RATELIMIT_COST, X_RATELIMIT_LIMIT, X_RATELIMIT_REMAINING, X_TASKS_QUEUED, X_TASKS_QUEUE_LIMIT

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from ApiResource

#requires

Methods included from DataSift

#build_path, request

Constructor Details

#initialize(config, stream) ⇒ LiveStream

Returns a new instance of LiveStream.



7
8
9
10
11
12
13
 
# File 'lib/live_stream.rb', line 7

def initialize(config, stream)
  super(config)
  @stream        = stream
  @retry_timeout = 0
  @subscriptions = {}
  @connected     = false
end

Instance Attribute Details

#connected ⇒ Object

Returns the value of attribute connected.



15
16
17
 
# File 'lib/live_stream.rb', line 15

def connected
  @connected
end

#retry_timeout ⇒ Object

Returns the value of attribute retry_timeout.



15
16
17
 
# File 'lib/live_stream.rb', line 15

def retry_timeout
  @retry_timeout
end

#stream ⇒ Object (readonly)

Returns the value of attribute stream.



15
16
17
 
# File 'lib/live_stream.rb', line 15

def stream
  @stream
end

#subscriptions ⇒ Object (readonly)

Returns the value of attribute subscriptions.



15
16
17
 
# File 'lib/live_stream.rb', line 15

def subscriptions
  @subscriptions
end

Instance Method Details

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


23
24
25
 
# File 'lib/live_stream.rb', line 23

def connected?
  @connected
end

#fire_ds_message(message) ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/live_stream.rb', line 27

def fire_ds_message(message)
  hash = false
  if message.has_key?(:hash)
    hash = message[:hash]
  end
  message.merge!({
    :is_failure => message[:status] == 'failure',
    :is_success => message[:status] == 'success',
    :is_warning => message[:status] == 'warning',
    :is_tick    => message[:status] == 'connected'
  })
  @on_datasift_message.call(self, message, hash)
end

#fire_on_message(hash, interaction) ⇒ Object 



41
42
43
44
45
46
47
 
# File 'lib/live_stream.rb', line 41

def fire_on_message(hash, interaction)
  callback = @subscriptions[hash]
  if callback == nil
    raise StreamingMessageError.new "no valid on_message callback provided for stream #{hash} with message #{interaction}"
  end
  callback.call(interaction, self, hash)
end

#on_datasift_message=(p) ⇒ Object

Raises:



18
19
20
21
 
# File 'lib/live_stream.rb', line 18

def on_datasift_message=(p)
  raise BadParametersError.new('on_ds_message - 3 parameters required') unless p.arity == 3
  @on_datasift_message = p
end

#subscribe(hash, on_message) ⇒ Object

Raises:



49
50
51
52
53
 
# File 'lib/live_stream.rb', line 49

def subscribe(hash, on_message)
  raise BadParametersError.new('on_message - 3 parameters required') unless on_message.arity == 3
  @subscriptions[hash] = on_message
  @stream.send "{ \"action\":\"subscribe\",\"hash\":\"#{hash}\"}"
end

#unsubscribe(hash) ⇒ Object 



55
56
57
 
# File 'lib/live_stream.rb', line 55

def unsubscribe hash
  @stream.send "{ \"action\":\"unsubscribe\",\"hash\":\"#{hash}\"}"
end