1 | package omq.server.remote.request; |
---|
2 | |
---|
3 | import java.util.ArrayList; |
---|
4 | import java.util.concurrent.BlockingQueue; |
---|
5 | import java.util.concurrent.LinkedBlockingDeque; |
---|
6 | |
---|
7 | import com.rabbitmq.client.QueueingConsumer; |
---|
8 | import com.rabbitmq.client.QueueingConsumer.Delivery; |
---|
9 | |
---|
10 | /** |
---|
11 | * |
---|
12 | * @author Sergi Toda <sergi.toda@estudiants.urv.cat> |
---|
13 | * |
---|
14 | */ |
---|
15 | public class RemoteWrapper { |
---|
16 | private RemoteObject obj; |
---|
17 | private int numThreads; |
---|
18 | private ArrayList<InvocationThread> invocationList; |
---|
19 | private BlockingQueue<Delivery> deliveryQueue; |
---|
20 | |
---|
21 | public RemoteWrapper(RemoteObject obj, int numThreads) { |
---|
22 | this.obj = obj; |
---|
23 | this.numThreads = numThreads; |
---|
24 | invocationList = new ArrayList<InvocationThread>(); |
---|
25 | deliveryQueue = new LinkedBlockingDeque<QueueingConsumer.Delivery>(); |
---|
26 | |
---|
27 | System.out.println("RemoteWrapper -> Object: " + obj.getRef() + ", numthreads listening = " + numThreads); |
---|
28 | |
---|
29 | for (int i = 0; i < numThreads; i++) { |
---|
30 | InvocationThread thread = new InvocationThread(obj, deliveryQueue); |
---|
31 | invocationList.add(thread); |
---|
32 | thread.start(); |
---|
33 | } |
---|
34 | } |
---|
35 | |
---|
36 | public void notifyDelivery(Delivery delivery) throws Exception { |
---|
37 | this.deliveryQueue.put(delivery); |
---|
38 | } |
---|
39 | |
---|
40 | public void stopRemoteWrapper() { |
---|
41 | for (InvocationThread thread : invocationList) { |
---|
42 | thread.interrupt(); |
---|
43 | } |
---|
44 | } |
---|
45 | |
---|
46 | public RemoteObject getObj() { |
---|
47 | return obj; |
---|
48 | } |
---|
49 | |
---|
50 | public void setObj(RemoteObject obj) { |
---|
51 | this.obj = obj; |
---|
52 | } |
---|
53 | |
---|
54 | public int getNumThreads() { |
---|
55 | return numThreads; |
---|
56 | } |
---|
57 | |
---|
58 | public void setNumThreads(int numThreads) { |
---|
59 | this.numThreads = numThreads; |
---|
60 | } |
---|
61 | |
---|
62 | public ArrayList<InvocationThread> getInvocationList() { |
---|
63 | return invocationList; |
---|
64 | } |
---|
65 | |
---|
66 | public void setInvocationList(ArrayList<InvocationThread> invocationList) { |
---|
67 | this.invocationList = invocationList; |
---|
68 | } |
---|
69 | |
---|
70 | public BlockingQueue<Delivery> getDeliveryQueue() { |
---|
71 | return deliveryQueue; |
---|
72 | } |
---|
73 | |
---|
74 | public void setDeliveryQueue(BlockingQueue<Delivery> deliveryQueue) { |
---|
75 | this.deliveryQueue = deliveryQueue; |
---|
76 | } |
---|
77 | } |
---|