Changeset 91 for branches/supervisor/src/main/java/omq/server/RemoteWrapper.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/RemoteWrapper.java (modified) (6 diffs)

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

@@ -24 +24 @@
         private RemoteObject obj;
         private int numThreads;
+        // private AtomicInteger busy;
+        private Object waitLock;
         private ArrayList<InvocationThread> invocationList;
         private BlockingQueue<Delivery> deliveryQueue;
@@ -30 +32 @@
                 this.obj = obj;
                 this.numThreads = numThreads;
+                // this.busy = new AtomicInteger(0);
+                this.waitLock = new Object();
                 invocationList = new ArrayList<InvocationThread>();
                 deliveryQueue = new LinkedBlockingDeque<QueueingConsumer.Delivery>();
@@ -36 +40 @@
 
                 for (int i = 0; i < numThreads; i++) {
-                        InvocationThread thread = new InvocationThread(obj, deliveryQueue, serializer);
+                        InvocationThread thread = new InvocationThread(obj, this, serializer);
                         invocationList.add(thread);
                         thread.start();
@@ -51 +55 @@
          */
         public void notifyDelivery(Delivery delivery) throws Exception {
+
+                // // Ensure there is at least one thread available
+                // while (this.busy.get() == this.numThreads) {
+                // System.out.println("Waiting for a thread available");
+                // logger.debug("Object reference: " + obj.getRef() + " is busy");
+                //
+                // synchronized (waitLock) {
+                // waitLock.wait();
+                // }
+                // }
+                // Notify an available thread
                 this.deliveryQueue.put(delivery);
+
         }
 
@@ -63 +79 @@
                 }
         }
+
+        // public int increaseBusy() {
+        // return this.busy.incrementAndGet();
+        // }
+        //
+        // public int decreaseBusy() {
+        // int value = this.busy.decrementAndGet();
+        // synchronized (waitLock) {
+        // waitLock.notifyAll();
+        // }
+        // return value;
+        // }
 
         public RemoteObject getObj() {
@@ -95 +123 @@
                 this.deliveryQueue = deliveryQueue;
         }
+
+        public Object getLock() {
+                return waitLock;
+        }
+
+        public void setLock(Object lock) {
+                this.waitLock = lock;
+        }
 }