parallel/services/RemoteEventProxy.java

package appl.parallel.services;

import java.rmi.RemoteException;
import java.util.Vector;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import appl.parallel.event.CommEventSink;
import appl.parallel.event.RemoteEvent;

/**
 * Delays events and fires them after the delay all <b>at once</b> in one
 * object (this will save communication time)
 * 
 * @author Dominik Appl
 */
public class RemoteEventProxy implements CommEventSink, Service {

        /**
         * This Thread submits the events after the delay
         * 
         * @author Dominik Appl
         */
        private class SubmitThread extends Thread {

                private boolean exit = false;

                private final ConcurrentLinkedQueue<RemoteEvent> queue2;

                private final CommEventSink sink;

                private final long delay;

                SubmitThread(ConcurrentLinkedQueue<RemoteEvent> queue,
                                CommEventSink sink, long delay) {
                        queue2 = queue;
                        this.sink = sink;
                        this.delay = Math.max(100, delay);
                }

                /*
                 * (non-Javadoc)
                 * 
                 * @see java.lang.Thread#run()
                 */
                public void run() {
                        while (!exit) {
                                try {
                                        this.sleep(delay);
                                } catch (InterruptedException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                }
                                // add all new events to a vector submit them at once
                                Vector<RemoteEvent> events = new Vector<RemoteEvent>();
                                while (queue.peek() != null)
                                        events.add(queue.poll());
                                try {
                                        if (events.size() > 0)
                                                sink.fireRemoteEvents((RemoteEvent[]) events
                                                                .toArray(new RemoteEvent[events.size()]));
                                } catch (RemoteException e) {
                                        // TODO Auto-generated catch block
                                        LOG.warn("Could not transmit events." + e.getMessage(), e);
                                        events.clear();
                                }
                        }
                }

                public void startThread() {
                        exit = false;
                        this.start();
                }

                public void stopThread() {
                        exit = true;
                }
        }

        private final Logger LOG = LogManager.getLogger(this.getClass().getName());

        private boolean timeMonitoring = false;

        private boolean transferMonitoring = false;

        private boolean isRunning;

        SubmitThread submitThread;

        ConcurrentLinkedQueue<RemoteEvent> queue;

        private final CommEventSink eventSink;

        private final long delay;

        /**
         * Creates a new proxy
         * 
         * @param targetSink
         *            the sink where the events should be submitted to
         * @param delay
         *            the delay (in milliseconds)
         */
        public RemoteEventProxy(CommEventSink targetSink, long delay) {
                this.eventSink = targetSink;
                this.delay = delay;
                queue = new ConcurrentLinkedQueue<RemoteEvent>();

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.event.RemoteEventSink#fireRemoteEvent(appl.parallel.event.RemoteEvent)
         */
        public void fireRemoteEvent(RemoteEvent e) throws RemoteException {
                queue.add(e);
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.event.RemoteEventSink#fireRemoteEvents(appl.parallel.event.RemoteEvent[])
         */
        public void fireRemoteEvents(RemoteEvent[] e) throws RemoteException {
                for (RemoteEvent event : e) {
                        fireRemoteEvent(event);
                }
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#isRunning()
         */
        public boolean isRunning() {
                return isRunning;
        }

        /**
         * @return true if the service is running and time monitoring is enabled
         */
        public boolean isTimeMonitoringEnabled() throws RemoteException {
                return timeMonitoring;
        }

        /**
         * @return true if the service is running and tansfer monitoring is enabled
         */
        public boolean isTransferMonitoringEnabled() throws RemoteException {
                return transferMonitoring;

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#startService()
         */
        public void startService() {
                isRunning = true;
                if (submitThread != null)
                        submitThread.stopThread();
                submitThread = new SubmitThread(queue, eventSink, delay);
                submitThread.startThread();
                try {
                        this.timeMonitoring = eventSink.isTimeMonitoringEnabled();
                        this.transferMonitoring = eventSink.isTransferMonitoringEnabled();
                } catch (RemoteException e) {
                        LOG.error("Could not connect to CommEventSink!!!");
                }

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#stopService()
         */
        public void stopService() {
                isRunning = false;
                if (submitThread != null)
                        submitThread.stopThread();
        }
}
1	package appl.parallel.services;
2
3	import java.rmi.RemoteException;
4	import java.util.Vector;
5	import java.util.concurrent.ConcurrentLinkedQueue;
6
7	import org.apache.log4j.LogManager;
8	import org.apache.log4j.Logger;
9
10	import appl.parallel.event.CommEventSink;
11	import appl.parallel.event.RemoteEvent;
12
13	/**
14	* Delays events and fires them after the delay all <b>at once</b> in one
15	* object (this will save communication time)
16	*
17	* @author Dominik Appl
18	*/
19	public class RemoteEventProxy implements CommEventSink, Service {
20
21	/**
22	* This Thread submits the events after the delay
23	*
24	* @author Dominik Appl
25	*/
26	private class SubmitThread extends Thread {
27
28	private boolean exit = false;
29
30	private final ConcurrentLinkedQueue<RemoteEvent> queue2;
31
32	private final CommEventSink sink;
33
34	private final long delay;
35
36	SubmitThread(ConcurrentLinkedQueue<RemoteEvent> queue,
37	CommEventSink sink, long delay) {
38	queue2 = queue;
39	this.sink = sink;
40	this.delay = Math.max(100, delay);
41	}
42
43	/*
44	* (non-Javadoc)
45	*
46	* @see java.lang.Thread#run()
47	*/
48	public void run() {
49	while (!exit) {
50	try {
51	this.sleep(delay);
52	} catch (InterruptedException e) {
53	// TODO Auto-generated catch block
54	e.printStackTrace();
55	}
56	// add all new events to a vector submit them at once
57	Vector<RemoteEvent> events = new Vector<RemoteEvent>();
58	while (queue.peek() != null)
59	events.add(queue.poll());
60	try {
61	if (events.size() > 0)
62	sink.fireRemoteEvents((RemoteEvent[]) events
63	.toArray(new RemoteEvent[events.size()]));
64	} catch (RemoteException e) {
65	// TODO Auto-generated catch block
66	LOG.warn("Could not transmit events." + e.getMessage(), e);
67	events.clear();
68	}
69	}
70	}
71
72	public void startThread() {
73	exit = false;
74	this.start();
75	}
76
77	public void stopThread() {
78	exit = true;
79	}
80	}
81
82	private final Logger LOG = LogManager.getLogger(this.getClass().getName());
83
84	private boolean timeMonitoring = false;
85
86	private boolean transferMonitoring = false;
87
88	private boolean isRunning;
89
90	SubmitThread submitThread;
91
92	ConcurrentLinkedQueue<RemoteEvent> queue;
93
94	private final CommEventSink eventSink;
95
96	private final long delay;
97
98	/**
99	* Creates a new proxy
100	*
101	* @param targetSink
102	* the sink where the events should be submitted to
103	* @param delay
104	* the delay (in milliseconds)
105	*/
106	public RemoteEventProxy(CommEventSink targetSink, long delay) {
107	this.eventSink = targetSink;
108	this.delay = delay;
109	queue = new ConcurrentLinkedQueue<RemoteEvent>();
110
111	}
112
113	/*
114	* (non-Javadoc)
115	*
116	* @see appl.parallel.event.RemoteEventSink#fireRemoteEvent(appl.parallel.event.RemoteEvent)
117	*/
118	public void fireRemoteEvent(RemoteEvent e) throws RemoteException {
119	queue.add(e);
120	}
121
122	/*
123	* (non-Javadoc)
124	*
125	* @see appl.parallel.event.RemoteEventSink#fireRemoteEvents(appl.parallel.event.RemoteEvent[])
126	*/
127	public void fireRemoteEvents(RemoteEvent[] e) throws RemoteException {
128	for (RemoteEvent event : e) {
129	fireRemoteEvent(event);
130	}
131	}
132
133	/*
134	* (non-Javadoc)
135	*
136	* @see appl.parallel.services.Service#isRunning()
137	*/
138	public boolean isRunning() {
139	return isRunning;
140	}
141
142	/**
143	* @return true if the service is running and time monitoring is enabled
144	*/
145	public boolean isTimeMonitoringEnabled() throws RemoteException {
146	return timeMonitoring;
147	}
148
149	/**
150	* @return true if the service is running and tansfer monitoring is enabled
151	*/
152	public boolean isTransferMonitoringEnabled() throws RemoteException {
153	return transferMonitoring;
154
155	}
156
157	/*
158	* (non-Javadoc)
159	*
160	* @see appl.parallel.services.Service#startService()
161	*/
162	public void startService() {
163	isRunning = true;
164	if (submitThread != null)
165	submitThread.stopThread();
166	submitThread = new SubmitThread(queue, eventSink, delay);
167	submitThread.startThread();
168	try {
169	this.timeMonitoring = eventSink.isTimeMonitoringEnabled();
170	this.transferMonitoring = eventSink.isTransferMonitoringEnabled();
171	} catch (RemoteException e) {
172	LOG.error("Could not connect to CommEventSink!!!");
173	}
174
175	}
176
177	/*
178	* (non-Javadoc)
179	*
180	* @see appl.parallel.services.Service#stopService()
181	*/
182	public void stopService() {
183	isRunning = false;
184	if (submitThread != null)
185	submitThread.stopThread();
186	}
187	}