svn commit: r1055968 - in /activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp: StompFrame.scala StompProtocolHandler.scala

[ch...@apache.org]

Author: chirino
Date: Thu Jan  6 17:58:15 2011
New Revision: 1055968

URL: http://svn.apache.org/viewvc?rev=1055968&view=rev
Log:
Let the client know what the broker authenticated them as when they connect.
Better disconnect handling, it will now reply if a receipt is present.

Modified:
    activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala
    activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala

Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala?rev=1055968&r1=1055967&r2=1055968&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala Thu Jan  6 17:58:15 2011
@@ -418,6 +418,7 @@ object Stomp {
 
   val BROWSER = ascii("browser")
   val EXCLUSIVE = ascii("exclusive")
+  val USER_ID = ascii("user-id")
 
   ///////////////////////////////////////////////////////////////////
   // Common Values

Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala?rev=1055968&r1=1055967&r2=1055968&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala Thu Jan  6 17:58:15 2011
@@ -438,7 +438,11 @@ class StompProtocolHandler extends Proto
               case CONNECT =>
                 on_stomp_connect(frame.headers)
               case DISCONNECT =>
-                connection.stop
+                send_receipt(frame.headers)
+                on_transport_disconnected
+                queue.after(die_delay, TimeUnit.MILLISECONDS) {
+                  connection.stop()
+                }
               case _ =>
                 die("Client must first send a connect frame");
             }
@@ -492,10 +496,6 @@ class StompProtocolHandler extends Proto
     connection.transport.resumeRead
   }
 
-  def weird(headers:HeaderMap) = {
-    println("weird: "+headers)
-  }
-
   def on_stomp_connect(headers:HeaderMap):Unit = {
 
     connection.transport match {
@@ -554,17 +554,20 @@ class StompProtocolHandler extends Proto
     def noop = shift {  k: (Unit=>Unit) => k() }
 
     def send_connected = {
+
+      var connected_headers = ListBuffer((VERSION, protocol_version))
+
+      session_id = encode_header(this.host.config.id + "-"+this.host.session_counter.incrementAndGet)
+      connected_headers += SESSION->session_id
+
       val outbound_heart_beat_header = ascii("%d,%d".format(outbound_heartbeat,inbound_heartbeat))
-      session_id = encode_header(this.host.config.id + ":"+this.host.session_counter.incrementAndGet)
-      if( connection_sink==null ) {
-        weird(headers)
-      }
-      connection_sink.offer(
-        StompFrame(CONNECTED, List(
-          (VERSION, protocol_version),
-          (SESSION, session_id),
-          (HEART_BEAT, outbound_heart_beat_header)
-        )))
+      connected_headers += HEART_BEAT->outbound_heart_beat_header
+
+      host.authenticator.user_name(security_context).foreach{ name=>
+        connected_headers += USER_ID->encode_header(name)
+      }
+
+      connection_sink.offer(StompFrame(CONNECTED,connected_headers.toList))
 
       if( this.host.direct_buffer_pool!=null ) {
         val wf = connection.transport.getProtocolCodec.asInstanceOf[StompCodec]