Class: RubyAMI::Stream

Inherits:
Object
  • Object
show all
Includes:
Celluloid::IO
Defined in:
lib/ruby_ami/stream.rb

Defined Under Namespace

Classes: ConnectionStatus

Constant Summary collapse

Connected = 
Class.new ConnectionStatus
Disconnected = 
Class.new ConnectionStatus

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(host, port, username, password, event_callback, logger = Logger, timeout = 0) ⇒ Stream

Returns a new instance of Stream.



21
22
23
24
25
26
27
28
 
# File 'lib/ruby_ami/stream.rb', line 21

def initialize(host, port, username, password, event_callback, logger = Logger, timeout = 0)
  super()
  @host, @port, @username, @password, @event_callback, @logger, @timeout = host, port, username, password, event_callback, logger, timeout
  logger.debug "Starting up..."
  @lexer = Lexer.new self
  @sent_actions   = {}
  @causal_actions = {}
end

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



17
18
19
 
# File 'lib/ruby_ami/stream.rb', line 17

def logger
  @logger
end

Instance Method Details

#message_received(message) ⇒ Object Also known as: error_received 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/ruby_ami/stream.rb', line 76

def message_received(message)
  logger.trace "[RECV] #{message.inspect}"
  case message
  when Event
    action = causal_action_for_event message
    if action
      action << message
      complete_causal_action_for_event message if action.complete?
    else
      fire_event message
    end
  when Response, Error
    action = sent_action_for_response message
    raise StandardError, "Received an AMI response with an unrecognized ActionID! #{message.inspect}" unless action
    action << message
  end
end

#post_init ⇒ Object 



50
51
52
53
54
 
# File 'lib/ruby_ami/stream.rb', line 50

def post_init
  @state = :started
  fire_event Connected.new
   @username, @password if @username && @password
end

#receive_data(data) ⇒ Object 



71
72
73
74
 
# File 'lib/ruby_ami/stream.rb', line 71

def receive_data(data)
  logger.trace "[RECV] #{data}"
  @lexer << data
end

#run ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/ruby_ami/stream.rb', line 34

def run
  Timeout::timeout(@timeout) do
    @socket = TCPSocket.from_ruby_socket ::TCPSocket.new(@host, @port)
  end
  post_init
  loop { receive_data @socket.readpartial(4096) }
rescue Errno::ECONNREFUSED, SocketError => e
  logger.error "Connection failed due to #{e.class}. Check your config and the server."
rescue EOFError
  logger.info "Client socket closed!"
rescue Timeout::Error
  logger.error "Timeout exceeded while trying to connect."
ensure
  async.terminate
end

#send_action(name, headers = {}) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
 
# File 'lib/ruby_ami/stream.rb', line 60

def send_action(name, headers = {})
  condition = Celluloid::Condition.new
  action = dispatch_action name, headers do |response|
    condition.signal response
  end
  condition.wait
  action.response.tap do |resp|
    abort resp if resp.is_a? Exception
  end
end

#send_data(data) ⇒ Object 



56
57
58
 
# File 'lib/ruby_ami/stream.rb', line 56

def send_data(data)
  @socket.write data
end

#syntax_error_encountered(ignored_chunk) ⇒ Object 



94
95
96
 
# File 'lib/ruby_ami/stream.rb', line 94

def syntax_error_encountered(ignored_chunk)
  logger.error "Encountered a syntax error. Ignoring chunk: #{ignored_chunk.inspect}"
end