Changeset 53 for trunk/src/main/java/omq/server

Timestamp:

06/20/13 16:57:39 (11 years ago)

Author:

stoda

Message:

Non static broker
TODO: change all test to see whether the new broker configuration works

Location:

trunk/src/main/java/omq/server

Files:

• InvocationThread.java (modified) (3 diffs)
• RemoteObject.java (modified) (5 diffs)
• RemoteWrapper.java (modified) (3 diffs)

trunk/src/main/java/omq/server/InvocationThread.java

@@ -23 +23 @@
         private static final Logger logger = Logger.getLogger(InvocationThread.class.getName());
         private RemoteObject obj;
+        private transient Serializer serializer;
         private BlockingQueue<Delivery> deliveryQueue;
         private boolean killed = false;
 
-        public InvocationThread(RemoteObject obj, BlockingQueue<Delivery> deliveryQueue) {
+        public InvocationThread(RemoteObject obj, BlockingQueue<Delivery> deliveryQueue, Serializer serializer) {
                 this.obj = obj;
                 this.deliveryQueue = deliveryQueue;
+                this.serializer = serializer;
         }
 
@@ -41 +43 @@
 
                                 // Deserialize the json
-                                Request request = Serializer.deserializeRequest(serializerType, delivery.getBody(), obj);
+                                Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj);
                                 // Log.saveLog("Server-Deserialize", delivery.getBody());
 
@@ -73 +75 @@
                                         BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build();
 
-                                        byte[] bytesResponse = Serializer.serialize(serializerType, resp);
+                                        byte[] bytesResponse = serializer.serialize(serializerType, resp);
                                         channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse);
 

trunk/src/main/java/omq/server/RemoteObject.java

@@ -39 +39 @@
         private String UID;
         private Properties env;
+        private transient Broker broker;
+        private transient Serializer serializer;
         private transient RemoteWrapper remoteWrapper;
         private transient Map<String, List<Class<?>>> params;
@@ -60 +62 @@
         }
 
-        public void startRemoteObject(String reference, Properties env) throws Exception {
-                this.UID = reference;
-                this.env = env;
+        public void startRemoteObject(String reference, Broker broker) throws Exception {
+                this.broker = broker;
+                UID = reference;
+                env = broker.getEnvironment();
+                serializer = broker.getSerializer();
 
                 params = new HashMap<String, List<Class<?>>>();
@@ -75 +79 @@
                 // Get num threads to use
                 int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1"));
-                remoteWrapper = new RemoteWrapper(this, numThreads);
+                remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer());
 
                 startQueues();
@@ -130 +134 @@
                 EventWrapper wrapper = new EventWrapper(event);
                 channel.exchangeDeclare(UID, "fanout");
-                channel.basicPublish(UID, "", null, Serializer.serialize(wrapper));
+                channel.basicPublish(UID, "", null, serializer.serialize(wrapper));
         }
 
@@ -221 +225 @@
 
                 // Start channel
-                channel = Broker.getNewChannel();
+                channel = broker.getNewChannel();
 
                 // Declares and bindings

trunk/src/main/java/omq/server/RemoteWrapper.java

@@ -4 +4 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
+
+import omq.common.util.Serializer;
 
 import org.apache.log4j.Logger;
@@ -23 +25 @@
         private BlockingQueue<Delivery> deliveryQueue;
 
-        public RemoteWrapper(RemoteObject obj, int numThreads) {
+        public RemoteWrapper(RemoteObject obj, int numThreads, Serializer serializer) {
                 this.obj = obj;
                 this.numThreads = numThreads;
@@ -32 +34 @@
 
                 for (int i = 0; i < numThreads; i++) {
-                        InvocationThread thread = new InvocationThread(obj, deliveryQueue);
+                        InvocationThread thread = new InvocationThread(obj, deliveryQueue, serializer);
                         invocationList.add(thread);
                         thread.start();