Changeset 91

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!!

Location:

branches/supervisor

Files:

• src/main/java/omq/common/broker/Broker.java (modified) (3 diffs)
• src/main/java/omq/common/broker/HasObject.java (added)
• src/main/java/omq/common/broker/RemoteBroker.java (added)
• src/main/java/omq/common/broker/RemoteBrokerImpl.java (added)
• src/main/java/omq/server/InvocationThread.java (modified) (4 diffs)
• src/main/java/omq/server/RemoteObject.java (modified) (2 diffs)
• src/main/java/omq/server/RemoteWrapper.java (modified) (6 diffs)
• src/main/java/omq/supervisor (added)
• src/main/java/omq/supervisor/ISupervisor.java (added)
• src/main/java/omq/supervisor/MultiBroker.java (added)
• src/main/java/omq/supervisor/OmqSettings.java (added)
• src/main/java/omq/supervisor/Supervisor.java (added)
• src/test/java/omq/test/lock (added)
• src/test/java/omq/test/lock/Sleep.java (added)
• src/test/java/omq/test/lock/SleepImpl.java (added)
• src/test/java/omq/test/lock/SleepTest.java (added)
• target/classes/log4j.xml (deleted)

branches/supervisor/src/main/java/omq/common/broker/Broker.java

@@ -49 +49 @@
         private boolean connectionClosed = false;
         private Properties environment = null;
+        private RemoteBrokerImpl remoteBrokerImpl;
         private Map<String, RemoteObject> remoteObjs;
         private Map<String, Object> proxies = new Hashtable<String, Object>();
@@ -357 +358 @@
         }
 
+        public void setSupervisor(String brokerSet, String brokerName) throws Exception {
+                remoteBrokerImpl = new RemoteBrokerImpl();
+                remoteBrokerImpl.startRemoteBroker(brokerSet, brokerName, this, getEnvironment());
+        }
+
         public Properties getEnvironment() {
                 return environment;
@@ -368 +374 @@
                 return serializer;
         }
+
+        public Map<String, RemoteObject> getRemoteObjs() {
+                return remoteObjs;
+        }
+
 }

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);

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

@@ -38 +38 @@
         private static final Logger logger = Logger.getLogger(RemoteObject.class.getName());
 
-        private String UID;
-        private Properties env;
-        private transient Broker broker;
-        private transient String multiQueue;
-        private transient RemoteWrapper remoteWrapper;
-        private transient Map<String, List<Class<?>>> params;
-        private transient Channel channel;
-        private transient QueueingConsumer consumer;
-        private transient boolean killed = false;
+        protected String UID;
+        protected Properties env;
+        protected transient Broker broker;
+        protected transient String multiQueue;
+        protected transient RemoteWrapper remoteWrapper;
+        protected transient Map<String, List<Class<?>>> params;
+        protected transient Channel channel;
+        protected transient QueueingConsumer consumer;
+        protected transient boolean killed = false;
 
         private static final Map<String, Class<?>> primitiveClasses = new HashMap<String, Class<?>>();
@@ -317 +317 @@
                  */
 
+                boolean autoAck = false;
+
                 // Declare a new consumer
                 consumer = new QueueingConsumer(channel);
-                channel.basicConsume(queue, true, consumer);
-                channel.basicConsume(multiQueue, true, consumer);
+                channel.basicConsume(queue, autoAck, consumer);
+                channel.basicConsume(multiQueue, autoAck, consumer);
         }
 

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;
+        }
 }