Module: EventMachine::Protocols::Stomp

Includes:
LineText2
Defined in:
lib/em/protocols/stomp.rb

Overview

Implements Stomp (docs.codehaus.org/display/STOMP/Protocol).

Usage example

module StompClient
  include EM::Protocols::Stomp

  def connection_completed
    connect :login => 'guest', :passcode => 'guest'
  end

  def receive_msg msg
    if msg.command == "CONNECTED"
      subscribe '/some/topic'
    else
      p ['got a message', msg]
      puts msg.body
    end
  end
end

EM.run{
  EM.connect 'localhost', 61613, StompClient
}

Defined Under Namespace

Classes: Message

Constant Summary

Constants included from LineText2

LineText2::MaxBinaryLength

Instance Method Summary collapse

Methods included from LineText2

#receive_data, #receive_end_of_binary_data, #set_binary_mode, #set_delimiter, #set_line_mode, #set_text_mode, #unbind

Instance Method Details

#ack(msgid) ⇒ Object

ACK command, for acknowledging receipt of messages

module StompClient
  include EM::P::Stomp

  def connection_completed
    connect :login => 'guest', :passcode => 'guest'
    # subscribe with ack mode
    subscribe '/some/topic', true
  end

  def receive_msg msg
    if msg.command == "MESSAGE"
      ack msg.headers['message-id']
      puts msg.body
    end
  end
end



198
199
200
 
# File 'lib/em/protocols/stomp.rb', line 198

def ack msgid
  send_frame "ACK", 'message-id'=> msgid
end

#connect(parms = {}) ⇒ Object

CONNECT command, for authentication

connect :login => 'guest', :passcode => 'guest'



159
160
161
 
# File 'lib/em/protocols/stomp.rb', line 159

def connect parms={}
  send_frame "CONNECT", parms
end

#init_message_reader ⇒ Object 



143
144
145
146
147
148
 
# File 'lib/em/protocols/stomp.rb', line 143

def init_message_reader
  @stomp_initialized = true
  set_delimiter "\n"
  set_line_mode
  @stomp_message = Message.new
end

#receive_binary_data(data) ⇒ Object 



136
137
138
139
140
 
# File 'lib/em/protocols/stomp.rb', line 136

def receive_binary_data data
  @stomp_message.body = data[0..-2]
  receive_msg(@stomp_message) if respond_to?(:receive_msg)
  init_message_reader
end

#receive_line(line) ⇒ Object 



121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/em/protocols/stomp.rb', line 121

def receive_line line
  @stomp_initialized || init_message_reader
  @stomp_message.consume_line(line) {|outcome|
    if outcome.first == :sized_text
      set_text_mode outcome[1]
    elsif outcome.first == :unsized_text
      set_delimiter "\0"
    elsif outcome.first == :dispatch
      receive_msg(@stomp_message) if respond_to?(:receive_msg)
      init_message_reader
    end
  }
end

#receive_msg(msg) ⇒ Object

Invoked with an incoming Stomp::Message received from the STOMP server



151
152
153
 
# File 'lib/em/protocols/stomp.rb', line 151

def receive_msg msg
  # stub, overwrite this in your handler
end

#send(destination, body, parms = {}) ⇒ Object

SEND command, for publishing messages to a topic

send '/topic/name', 'some message here'



167
168
169
 
# File 'lib/em/protocols/stomp.rb', line 167

def send destination, body, parms={}
  send_frame "SEND", parms.merge( :destination=>destination ), body.to_s
end

#send_frame(verb, headers = {}, body = "") ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/em/protocols/stomp.rb', line 106

def send_frame verb, headers={}, body=""
  body = body.to_s
  ary = [verb, "\n"]
  body_bytesize = body.bytesize if body.respond_to? :bytesize
  body_bytesize ||= body.size
  headers.each {|k,v| ary << "#{k}:#{v}\n" }
  ary << "content-length: #{body_bytesize}\n"
  ary << "content-type: text/plain; charset=UTF-8\n" unless headers.has_key? 'content-type'
  ary << "\n"
  ary << body
  ary << "\0"
  send_data ary.join
end

#subscribe(dest, ack = false) ⇒ Object

SUBSCRIBE command, for subscribing to topics

subscribe '/topic/name', false



175
176
177
 
# File 'lib/em/protocols/stomp.rb', line 175

def subscribe dest, ack=false
  send_frame "SUBSCRIBE", {:destination=>dest, :ack=>(ack ? "client" : "auto")}
end