amq stomp error

gistfile1.txt

I, [2012-05-01T11:53:03.493289 #3624]  INFO -- : Logger initialization complete.
D, [2012-05-01T11:53:03.494289 #3624] DEBUG -- : Connecting: Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:03.501290 #3624] DEBUG -- : Connected: Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
I, [2012-05-01T11:53:03.501290 #3624]  INFO -- : Using STOMP gem Version: 1.2.2
I, [2012-05-01T11:53:03.502290 #3624]  INFO -- : Subscribing to /queue/font_request
D, [2012-05-01T11:53:03.502290 #3624] DEBUG -- : Subscribe Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:03.502290 #3624] DEBUG -- : Subscribe Headers destination/queue/font_requestackclient
D, [2012-05-01T11:53:08.218762 #3624] DEBUG -- : Receive Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.218762 #3624] DEBUG -- : Receive Result <Stomp::Message headers={"expires"=>"1335211972613", "timestamp"=>"1335211971613", "destination"=>"/queue/font_request", "message-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1", "content-type"=>"text/plain; charset=UTF-8", "priority"=>"4", "content-length"=>"1"} body=' ' command='MESSAGE' >
I, [2012-05-01T11:53:08.218762 #3624]  INFO -- : Received: {"expires"=>"1335211972613", "timestamp"=>"1335211971613", "destination"=>"/queue/font_request", "message-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1", "content-type"=>"text/plain; charset=UTF-8", "priority"=>"4", "content-length"=>"1"}
Content Length: 1
I, [2012-05-01T11:53:08.222762 #3624]  INFO -- : Sending response: {"ps_destination"=>"/queue/font_request", "ps_message-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1", "ps_content-type"=>"text/plain; charset=UTF-8", "expires"=>"1335211972613", "priority"=>"4", "content-length"=>"1", "ps_timestamp"=>"1335211971613", "correlation-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1"} to test_response
D, [2012-05-01T11:53:08.222762 #3624] DEBUG -- : Publish Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.222762 #3624] DEBUG -- : Publish Message 
D, [2012-05-01T11:53:08.223762 #3624] DEBUG -- : Publish Headers destination/queue/test_responseps_timestamp1335211971613expires1335211972613ps_message-idID:martha.centos.com-41813-1335211741474-2:23:-1:1:1ps_content-typetext/plain; charset=UTF-8ps_destination/queue/font_requestcontent-length1correlation-idID:martha.centos.com-41813-1335211741474-2:23:-1:1:1priority4
I, [2012-05-01T11:53:08.234763 #3624]  INFO -- : Sending Ack for: ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1
D, [2012-05-01T11:53:08.240764 #3624] DEBUG -- : Miscellaneous Error Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.240764 #3624] DEBUG -- : Miscellaneous Error String transmit to 10.70.31.39 failed: An established connection was aborted by the software in your host machine.

D, [2012-05-01T11:53:08.240764 #3624] DEBUG -- : Connecting: Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.290769 #3624] DEBUG -- : Connected: Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.352775 #3624] DEBUG -- : Receive Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.353775 #3624] DEBUG -- : Receive Result <Stomp::Message headers={"expires"=>"1335211972613", "timestamp"=>"1335211971613", "destination"=>"/queue/font_request", "message-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1", "content-type"=>"text/plain; charset=UTF-8", "priority"=>"4", "content-length"=>"1", "redelivered"=>"true"} body=' ' command='MESSAGE' >
I, [2012-05-01T11:53:08.354775 #3624]  INFO -- : Received: {"expires"=>"1335211972613", "timestamp"=>"1335211971613", "destination"=>"/queue/font_request", "message-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1", "content-type"=>"text/plain; charset=UTF-8", "priority"=>"4", "content-length"=>"1", "redelivered"=>"true"}
Content Length: 1
I, [2012-05-01T11:53:08.355775 #3624]  INFO -- : Sending response: {"ps_destination"=>"/queue/font_request", "ps_message-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1", "ps_content-type"=>"text/plain; charset=UTF-8", "ps_redelivered"=>"true", "expires"=>"1335211972613", "priority"=>"4", "content-length"=>"1", "ps_timestamp"=>"1335211971613", "correlation-id"=>"ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1"} to test_response
D, [2012-05-01T11:53:08.356776 #3624] DEBUG -- : Publish Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.357776 #3624] DEBUG -- : Publish Message 
D, [2012-05-01T11:53:08.357776 #3624] DEBUG -- : Publish Headers destination/queue/test_responseps_timestamp1335211971613expires1335211972613ps_message-idID:martha.centos.com-41813-1335211741474-2:23:-1:1:1ps_content-typetext/plain; charset=UTF-8ps_destination/queue/font_requestps_redeliveredtruecontent-length1correlation-idID:martha.centos.com-41813-1335211741474-2:23:-1:1:1priority4
I, [2012-05-01T11:53:08.382778 #3624]  INFO -- : Sending Ack for: ID:martha.centos.com-41813-1335211741474-2:23:-1:1:1
D, [2012-05-01T11:53:08.400780 #3624] DEBUG -- : Receive Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.401780 #3624] DEBUG -- : Receive Result <Stomp::Message headers={"message"=>"content-length bytes were read and there was no trailing null byte"} body='org.apache.activemq.transport.stomp.ProtocolException: content-length bytes were read and there was no trailing null byte
	at org.apache.activemq.transport.stomp.StompWireFormat.unmarshal(StompWireFormat.java:109)
	at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:227)
	at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:219)
	at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:202)
	at java.lang.Thread.run(Unknown Source)
' command='ERROR' >
I, [2012-05-01T11:53:08.402780 #3624]  INFO -- : Received: {"message"=>"content-length bytes were read and there was no trailing null byte"}
W, [2012-05-01T11:53:08.402780 #3624]  WARN -- : ERROR frame received: <Stomp::Message headers={"message"=>"content-length bytes were read and there was no trailing null byte"} body='org.apache.activemq.transport.stomp.ProtocolException: content-length bytes were read and there was no trailing null byte
	at org.apache.activemq.transport.stomp.StompWireFormat.unmarshal(StompWireFormat.java:109)
	at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:227)
	at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:219)
	at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:202)
	at java.lang.Thread.run(Unknown Source)
' command='ERROR' >
D, [2012-05-01T11:53:08.403780 #3624] DEBUG -- : Receive Parms Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.404780 #3624] DEBUG -- : Receive Result <Stomp::Message headers={"message"=>"Unknown STOMP action: CK"} body='org.apache.activemq.transport.stomp.ProtocolException: Unknown STOMP action: CK
	at org.apache.activemq.transport.stomp.ProtocolConverter.onStompCommand(ProtocolConverter.java:196)
	at org.apache.activemq.transport.stomp.StompTransportFilter.onCommand(StompTransportFilter.java:70)
	at org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
	at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:220)
	at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:202)
	at java.lang.Thread.run(Unknown Source)
' command='ERROR' >
I, [2012-05-01T11:53:08.405780 #3624]  INFO -- : Received: {"message"=>"Unknown STOMP action: CK"}
W, [2012-05-01T11:53:08.405780 #3624]  WARN -- : ERROR frame received: <Stomp::Message headers={"message"=>"Unknown STOMP action: CK"} body='org.apache.activemq.transport.stomp.ProtocolException: Unknown STOMP action: CK
	at org.apache.activemq.transport.stomp.ProtocolConverter.onStompCommand(ProtocolConverter.java:196)
	at org.apache.activemq.transport.stomp.StompTransportFilter.onCommand(StompTransportFilter.java:70)
	at org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
	at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:220)
	at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:202)
	at java.lang.Thread.run(Unknown Source)
' command='ERROR' >
D, [2012-05-01T11:53:08.406781 #3624] DEBUG -- : Miscellaneous Error Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.407781 #3624] DEBUG -- : Miscellaneous Error String connection.receive returning EOF as nil - resetting connection.

D, [2012-05-01T11:53:08.407781 #3624] DEBUG -- : Connecting: Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false
D, [2012-05-01T11:53:08.423782 #3624] DEBUG -- : Connected: Host: 10.70.31.39, Port: 61613, Login: Port: , Passcode: , ssl: false

receiver.rb

require 'stomp'
require 'logger'
require 'slogger'

stomp_logger = Slogger::new

@logger = Logger.new(STDOUT)
@queue_name = "/queue/font_request"
@subscription_headers = {:ack => 'client'}

stomp_connection_hash = {
    :hosts => [
        {:login => '',
         :passcode => '',
         :host => "10.70.31.39",
         :port => 61613,
         :ssl => false}

    ],
    :initial_reconnect_delay => 5,
    :use_exponential_back_off => false,
    :logger => stomp_logger,
}


class Request
  attr_reader :id, :service, :headers, :body
  def initialize(id, service, headers = {}, body = '')
    @id, @service, @headers, @body = id, service, headers, body
  end
end

class Response
  attr_accessor :headers, :body
  def initialize(headers, body)
    @headers = headers.nil? ? {} : headers
    @body = body
  end
end

def prepare_headers(headers)
  new_headers = {}

  # See http://activemq.apache.org/stomp.html for more information on STOMP headers
  # automatically mapping to JMS Message Properties
  stomp_to_JMS_keywords = ["correlation-id", "expires", "persistent",
                           "priority", "reply-to", "type", "JMSXGroupID", "JMSXGroupSeq", "content-length"]
  headers.each do |k, v|
    if stomp_to_JMS_keywords.include?(k.to_s)
      new_headers["#{k}"] = v.to_s.gsub("\n", "\\n").gsub("\r", '')
    else
      # Adding a prefix to all of our headers so we can determine which headers
      # are our own and which are native headers
      new_headers["ps_#{k}"] = v.to_s.gsub("\n", "\\n").gsub("\r", '')
    end
  end
  new_headers
end

def send_response(request, response)
  body = response.body
  headers = prepare_headers(response.headers).merge('correlation-id' => request.id)
  response_queue = "#{request.service}_response"
  @logger.info "Sending response: #{headers.inspect} to #{response_queue}"
  @conn.publish("/queue/#{response_queue}", body, headers)
end

@conn = Stomp::Connection.new(stomp_connection_hash)
@logger.info "Using STOMP gem Version: #{Stomp::Version::STRING}"
@logger.info "Subscribing to #{@queue_name}"
@conn.subscribe(@queue_name, @subscription_headers)


loop do
  begin
    msg = @conn.receive
    @logger.info "Received: #{msg.headers.inspect}"

    if msg.command == 'ERROR'
      @logger.warn "ERROR frame received: #{msg}"
      next
    elsif msg.command != 'MESSAGE'
      @logger.warn "Unexpected message received: #{msg}"
      next
    end

    request = Request.new(msg.headers['message-id'], "test", msg.headers, msg.body)
    response = Response.new(request.headers, '')
    send_response(request, response)                 
    @logger.info "Sending Ack for: #{msg.headers['message-id']}"

    @conn.ack(msg.headers['message-id'])

  rescue SystemExit, StopServer, Errno::EBADF
    break
  rescue Exception => e
    @logger.error(e)
  end
end

sender.rb

require 'stomp'

headers = {}
body = "\x00"

client = Stomp::Client.new("", "", "10.70.31.39", 61613)
client.publish("/queue/font_request", body, headers)

puts "message sent"