svn commit: r1419818 - in /activemq/activemq-apollo/trunk/apollo-openwire/src: main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala test/scala/org/apache/activemq/apollo/openwire/test/OpenwireParallelTest.scala

[ch...@apache.org]

Author: chirino
Date: Mon Dec 10 21:48:51 2012
New Revision: 1419818

URL: http://svn.apache.org/viewvc?rev=1419818&view=rev
Log:
Fixes APLO-277: Openwire durable sub delete hangs / fails.

Modified:
    activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
    activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/test/OpenwireParallelTest.scala

Modified: activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala?rev=1419818&r1=1419817&r2=1419818&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala Mon Dec 10 21:48:51 2012
@@ -281,6 +281,7 @@ class OpenwireProtocolHandler extends Pr
           case info:ShutdownInfo=> ack(info); connection.stop(NOOP)
           case info:FlushCommand=> ack(info)
           case info:DestinationInfo=> on_destination_info(info)
+          case info:RemoveSubscriptionInfo=> on_remove_subscription_info(info)
 
           // case info:ConnectionControl=>
           // case info:ConnectionError=>
@@ -510,6 +511,27 @@ class OpenwireProtocolHandler extends Pr
     }
   }
 
+  def on_remove_subscription_info(info: RemoveSubscriptionInfo) ={
+    var subscription_id = ""
+    if( info.getClientId != null ) {
+      subscription_id += info.getClientId + ":"
+    }
+    subscription_id += info.getSubscriptionName
+    host.dispatch_queue {
+      host.local_router.dsub_domain.destination_by_id.get(subscription_id) match {
+        case None =>
+          queue(fail("The subscription does not exist", info))
+        case Some(dest:Queue) =>
+          host.local_router._destroy_queue(dest, security_context) match {
+            case Some(error) =>
+              queue(fail(error, info))
+            case None=>
+              queue(ack(info))
+          }
+      }
+    }
+  }
+
   def on_remove_info(info: RemoveInfo) = {
     info.getObjectId match {
       case id: ConnectionId =>

Modified: activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/test/OpenwireParallelTest.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/test/OpenwireParallelTest.scala?rev=1419818&r1=1419817&r2=1419818&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/test/OpenwireParallelTest.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/test/OpenwireParallelTest.scala Mon Dec 10 21:48:51 2012
@@ -721,4 +721,16 @@ class OpenwireParallelTest extends Openw
     consumer.receive(1000) should equal(outbound(2))
   }
 
+  test("Unsubscribe invalid durable sub") {
+    connect()
+    val session = default_connection.createSession(false, Session.AUTO_ACKNOWLEDGE)
+    try {
+      session.unsubscribe("does not exist")
+      fail("Expecting JMS Exception")
+    } catch {
+      case e:JMSException =>
+      case _ => fail("Expecting JMS Exception")
+    }
+ }
+
 }
\ No newline at end of file