Merge pull request #55 from iitsoftware/vt

Vt

Author: iitsoftware

pom.xml

@@ -7,7 +7,7 @@
 
   <groupId>com.swiftmq</groupId>
   <artifactId>swiftmq-client</artifactId>
-  <version>12.5.4</version>
+  <version>13.0.0</version>
 
   <name>SwiftMQ Client</name>
   <description>Client for SwiftMQ Messaging System with JMS, AMQP 1.0 and file transfer over JMS.</description>

src/main/java/com/swiftmq/admin/cli/CLI.java

@@ -17,6 +17,7 @@
 
 package com.swiftmq.admin.mgmt;
 
+import com.swiftmq.admin.mgmt.v750.EndpointImpl;
 import com.swiftmq.jms.BytesMessageImpl;
 import com.swiftmq.jms.MessageImpl;
 import com.swiftmq.jms.QueueImpl;
@@ -103,29 +104,19 @@ public Endpoint create(RequestServiceFactory rsf, boolean createInternalCommands
 
             Endpoint endpoint = null;
             try {
-                switch (MGMT_PROTOCOL_VERSION) {
-                    case 750: {
-                        ProtocolReply pr = (ProtocolReply) request(new ProtocolRequest(750));
-                        if (pr.isOk()) {
-                            endpoint = new com.swiftmq.admin.mgmt.v750.EndpointImpl(connection, senderSession, sender, receiverSession, receiver, replyQueue, rsf.createRequestService(750), createInternalCommands);
-                            endpoint.setSubscriptionFilterEnabled(true);
-                        } else {
-                            pr = (ProtocolReply) request(new ProtocolRequest(400));
-                            if (!pr.isOk())
-                                throw pr.getException();
-                            endpoint = new com.swiftmq.admin.mgmt.v400.EndpointImpl(connection, senderSession, sender, receiverSession, receiver, replyQueue, rsf.createRequestService(400), createInternalCommands);
-                        }
-                    }
-                    break;
-                    case 400: {
-                        ProtocolReply pr = (ProtocolReply) request(new ProtocolRequest(400));
+                if (MGMT_PROTOCOL_VERSION == 750) {
+                    ProtocolReply pr = (ProtocolReply) request(new ProtocolRequest(750));
+                    if (pr.isOk()) {
+                        endpoint = new EndpointImpl(connection, senderSession, sender, receiverSession, receiver, replyQueue, rsf.createRequestService(750), createInternalCommands);
+                        endpoint.setSubscriptionFilterEnabled(true);
+                    } else {
+                        pr = (ProtocolReply) request(new ProtocolRequest(400));
                         if (!pr.isOk())
                             throw pr.getException();
-                        endpoint = new com.swiftmq.admin.mgmt.v400.EndpointImpl(connection, senderSession, sender, receiverSession, receiver, replyQueue, rsf.createRequestService(400), createInternalCommands);
+                        endpoint = new com.swiftmq.admin.mgmt.v750.EndpointImpl(connection, senderSession, sender, receiverSession, receiver, replyQueue, rsf.createRequestService(400), createInternalCommands);
                     }
-                    break;
-                    default:
-                        throw new Exception("Invalid management protocol version (set via swiftmq.mgmt.protocol.version): " + MGMT_PROTOCOL_VERSION);
+                } else {
+                    throw new Exception("Invalid management protocol version (set via swiftmq.mgmt.protocol.version): " + MGMT_PROTOCOL_VERSION);
                 }
             } catch (Exception e) {
                 cleanup();

src/main/java/com/swiftmq/admin/cli/v400/RequestProcessor.java

@@ -17,41 +17,36 @@
 
 package com.swiftmq.admin.mgmt;
 
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class EndpointRegistry {
-    Map endpoints = new HashMap();
-    boolean closed = false;
+    private final Map<String, Endpoint> endpoints = new ConcurrentHashMap<>();
+    private volatile boolean closed = false;
 
     public EndpointRegistry() {
     }
 
-    public synchronized void put(String routerName, Endpoint endpoint) throws EndpointRegistryClosedException {
+    public void put(String routerName, Endpoint endpoint) throws EndpointRegistryClosedException {
         if (closed)
             throw new EndpointRegistryClosedException("EndpointRegistry already closed!");
         endpoints.put(routerName, endpoint);
     }
 
-    public synchronized Endpoint get(String routerName) {
+    public Endpoint get(String routerName) {
         return (Endpoint) endpoints.get(routerName);
     }
 
-    public synchronized Endpoint remove(String routerName) {
+    public Endpoint remove(String routerName) {
         return (Endpoint) endpoints.remove(routerName);
     }
 
     public void close() {
-        Map map;
-        synchronized (this) {
-            map = (Map) ((HashMap) endpoints).clone();
-            endpoints.clear();
-            closed = true;
-        }
-        for (Iterator iter = map.entrySet().iterator(); iter.hasNext(); ) {
-            Endpoint endpoint = (Endpoint) ((Map.Entry) iter.next()).getValue();
+        for (Map.Entry<String, Endpoint> o : endpoints.entrySet()) {
+            Endpoint endpoint = o.getValue();
             endpoint.close();
         }
+        endpoints.clear();
+        closed = true;
     }
 }