Class: Immunio::Channel

Inherits:
Object
  • Object
show all
Defined in:
lib/immunio/channel.rb

Overview

Communication channel with the Immunio webservice.

Constant Summary collapse

DIGEST = 
OpenSSL::Digest.new('sha1')

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ Channel

Returns a new instance of Channel.



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/immunio/channel.rb', line 19

def initialize(config)
  @config = config

  @agent_uuid = nil

  # Messages waiting to be sent.
  @message_queue = Queue.new
  # Messages that were sent but failed. Need to be resent.
  @send_buffer = []
  @send_buffer_bytes = 0
  @last_report_time = 0

  # A large message we may have popped from the queue but couldn't fit
  # in the current report
  @next_message = nil

  @send_seq = 0
  @dropped_message_count = 0
  @rejected_message_count = 0
  @success_count = 0
  @error_count = 0
  @quick_connect = true

  @started = false
  @ready = false

  @callbacks = []

  # Anything looking to add to the messages sent to the server:
  @senders = []
end

Instance Attribute Details

#error_count ⇒ Object (readonly)

Returns the value of attribute error_count.



16
17
18
 
# File 'lib/immunio/channel.rb', line 16

def error_count
  @error_count
end

#message_queue ⇒ Object (readonly)

Returns the value of attribute message_queue.



16
17
18
 
# File 'lib/immunio/channel.rb', line 16

def message_queue
  @message_queue
end

#rejected_message_count ⇒ Object (readonly)

Returns the value of attribute rejected_message_count.



17
18
19
 
# File 'lib/immunio/channel.rb', line 17

def rejected_message_count
  @rejected_message_count
end

#success_count ⇒ Object (readonly)

Returns the value of attribute success_count.



16
17
18
 
# File 'lib/immunio/channel.rb', line 16

def success_count
  @success_count
end

Instance Method Details

#messages_count ⇒ Object 



63
64
65
 
# File 'lib/immunio/channel.rb', line 63

def messages_count
  @message_queue.size
end

#on_message(&block) ⇒ Object 



107
108
109
 
# File 'lib/immunio/channel.rb', line 107

def on_message(&block)
  @callbacks << block
end

#on_sending(&block) ⇒ Object 



111
112
113
 
# File 'lib/immunio/channel.rb', line 111

def on_sending(&block)
  @senders << block
end

#ready? ⇒ Boolean

Returns:

  • (Boolean) 


51
52
53
 
# File 'lib/immunio/channel.rb', line 51

def ready?
  @ready
end

#send_encoded_message(message) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/immunio/channel.rb', line 93

def send_encoded_message(message)
  if @message_queue.size > @config.max_send_queue_size
    Immunio.logger.warn { "Dropping message for agent manager due to queue overflow (#{@message_queue.size} > #{@config.max_send_queue_size})" }
    Immunio.logger.debug { "Dropped message: (#{message})" }
    # No room for this message on the queue. Discard.
    @dropped_message_count += 1
    return
  end

  Immunio.logger.debug {"Sending message to backend: #{MessagePack.unpack(message)}"}

  @message_queue << message
end

#send_message(message) ⇒ Object 



89
90
91
 
# File 'lib/immunio/channel.rb', line 89

def send_message(message)
  send_encoded_message message.to_msgpack
end

#set_ready ⇒ Object 



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

def set_ready
  @ready = true
end

#start ⇒ Object 



67
68
69
70
71
72
 
# File 'lib/immunio/channel.rb', line 67

def start
  return if @started

  @started = true
  @thread = Thread.new { run }
end

#started? ⇒ Boolean

Returns:

  • (Boolean) 


59
60
61
 
# File 'lib/immunio/channel.rb', line 59

def started?
  @started
end

#stop ⇒ Object

Stop and wait for the last messages to be sent.



75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/immunio/channel.rb', line 75

def stop
  return unless @started

  Immunio.logger.debug { "Stopping channel" }

  @started = false
  @ready = false

  if @thread
    @thread.kill
    @thread.join
  end
end

#wait_until_ready! ⇒ Object

Wait until we receive a message from the agentmanager. This is used primarily for internal testing to wait until all the hooks are loaded.



118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/immunio/channel.rb', line 118

def wait_until_ready!
  return if @ready

  if @config.ready_timeout.to_i <= 0
    return
  end

  Immunio.logger.debug { "Channel waiting #{@config.ready_timeout.to_i} seconds until ready..." }
  Timeout.timeout @config.ready_timeout.to_i do
    # Wait until we get a response from the agentmanager
    sleep 0.1 until ready?
    Immunio.logger.debug { "Channel ready!" }
  end
end