Changeset 91 for branches/supervisor/src/main/java/omq/server/InvocationThread.java

Timestamp:

09/27/13 17:50:41 (11 years ago)

Author:

stoda

Message:

Semaphores added and removed, ack error discovered and solutioned... Some tests added

Supervisor interface created and more things I'll do later...

TODO: supervisor!!

File:

• branches/supervisor/src/main/java/omq/server/InvocationThread.java (modified) (4 diffs)

branches/supervisor/src/main/java/omq/server/InvocationThread.java

@@ -24 +24 @@
         private static final Logger logger = Logger.getLogger(InvocationThread.class.getName());
         private RemoteObject obj;
-        private transient Serializer serializer;
+        private Serializer serializer;
+        // private RemoteWrapper wrapper;
         private BlockingQueue<Delivery> deliveryQueue;
         private boolean killed = false;
 
-        public InvocationThread(RemoteObject obj, BlockingQueue<Delivery> deliveryQueue, Serializer serializer) {
+        public InvocationThread(RemoteObject obj, RemoteWrapper wrapper, Serializer serializer) {
                 this.obj = obj;
-                this.deliveryQueue = deliveryQueue;
+                // this.wrapper = wrapper;
+                this.deliveryQueue = wrapper.getDeliveryQueue();
                 this.serializer = serializer;
         }
@@ -40 +42 @@
                                 // Get the delivery
                                 Delivery delivery = deliveryQueue.take();
+
+                                // // Indicate this thread is not available
+                                // wrapper.increaseBusy();
 
                                 String serializerType = delivery.getProperties().getType();
@@ -64 +69 @@
                                 }
 
+                                
+                                Channel channel = obj.getChannel();
+                                
+                                
                                 // Reply if it's necessary
                                 if (!request.isAsync()) {
                                         Response resp = new Response(request.getId(), obj.getRef(), result, error);
 
-                                        Channel channel = obj.getChannel();
+                                        
 
                                         BasicProperties props = delivery.getProperties();
@@ -79 +88 @@
                                                         + props.getReplyTo());
                                 }
-
+                                
+                                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+                                
+                                // // Indicate this thread is available
+                                // wrapper.decreaseBusy();
                         } catch (InterruptedException i) {
                                 logger.error(i);