Class: Carrot::AMQP::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/carrot/amqp/server.rb

Defined Under Namespace

Classes: ProtocolError, ServerDown

Constant Summary collapse

CONNECT_TIMEOUT = 
1.0
RETRY_DELAY = 
10.0
DEFAULT_PORT = 
5672

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(opts = {}) ⇒ Server

Returns a new instance of Server.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/carrot/amqp/server.rb', line 23

def initialize(opts = {})
  @host   = opts[:host]  || 'localhost'
  @port   = opts[:port]  || DEFAULT_PORT
  @user   = opts[:user]  || 'guest'
  @pass   = opts[:pass]  || 'guest'
  @vhost  = opts[:vhost] || '/'
  @insist = opts[:insist]
  @status = 'NOT CONNECTED'

  @use_ssl    = !!opts[:ssl]
  @ssl_verify = opts[:ssl_verify] || OpenSSL::SSL::VERIFY_PEER

  @multithread = opts[:multithread]      
  start_session
end

Instance Attribute Details

#channel ⇒ Object

Returns the value of attribute channel.



18
19
20
 
# File 'lib/carrot/amqp/server.rb', line 18

def channel
  @channel
end

#host ⇒ Object (readonly)

Returns the value of attribute host.



17
18
19
 
# File 'lib/carrot/amqp/server.rb', line 17

def host
  @host
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



17
18
19
 
# File 'lib/carrot/amqp/server.rb', line 17

def port
  @port
end

#status ⇒ Object (readonly)

Returns the value of attribute status.



17
18
19
 
# File 'lib/carrot/amqp/server.rb', line 17

def status
  @status
end

#ticket ⇒ Object

Returns the value of attribute ticket.



18
19
20
 
# File 'lib/carrot/amqp/server.rb', line 18

def ticket
  @ticket
end

Instance Method Details

#close ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/carrot/amqp/server.rb', line 66

def close
  send_frame(
    Protocol::Channel::Close.new(:reply_code => 200, :reply_text => 'bye', :method_id => 0, :class_id => 0)
  )
  puts "Error closing channel #{channel}" unless next_method.is_a?(Protocol::Channel::CloseOk)

  self.channel = 0
  send_frame(
    Protocol::Connection::Close.new(:reply_code => 200, :reply_text => 'Goodbye', :class_id => 0, :method_id => 0)
  )
  puts "Error closing connection" unless next_method.is_a?(Protocol::Connection::CloseOk)

rescue ServerDown => e
ensure
  close_socket
end

#next_frame ⇒ Object 



51
52
53
54
55
 
# File 'lib/carrot/amqp/server.rb', line 51

def next_frame
  frame = Frame.parse(buffer)
  log :received, frame
  frame
end

#next_method ⇒ Object 



57
58
59
 
# File 'lib/carrot/amqp/server.rb', line 57

def next_method
  next_payload
end

#next_payload ⇒ Object 



61
62
63
64
 
# File 'lib/carrot/amqp/server.rb', line 61

def next_payload
  frame = next_frame
  frame and frame.payload
end

#read(*args) ⇒ Object 



83
84
85
 
# File 'lib/carrot/amqp/server.rb', line 83

def read(*args)
  send_command(:read, *args)
end

#send_frame(*args) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/carrot/amqp/server.rb', line 39

def send_frame(*args)
  args.each do |data|
    data.ticket  = ticket if ticket and data.respond_to?(:ticket=)
    data         = data.to_frame(channel) unless data.is_a?(Frame)
    data.channel = channel

    log :send, data
    write(data.to_s)
  end
  nil
end

#write(*args) ⇒ Object 



87
88
89
 
# File 'lib/carrot/amqp/server.rb', line 87

def write(*args)
  send_command(:write, *args)
end