avoid synchronization on boxed primitive and use a mutex instead
*/
package org.wso2.siddhi.core.stream.input;
+import org.apache.log4j.Logger;
import org.wso2.siddhi.core.config.ExecutionPlanContext;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.util.snapshot.Snapshotable;
@@ -25,32 +26,41 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.Semaphore;
/**
* This class wraps {@link InputHandler} class in order to guarantee exactly once processing
*/
public class InputEventHandler implements Snapshotable {
+ private static final Logger log = Logger.getLogger(InputEventHandler.class);
private InputHandler inputHandler;
private String elementId;
/**
* {@link Event#id} of the last event which is processed by this mapper
*/
private Long lastEventId = null;
+ private final Semaphore mutex;
public InputEventHandler(InputHandler inputHandler, ExecutionPlanContext executionPlanContext) {
this.inputHandler = inputHandler;
this.elementId = executionPlanContext.getElementIdGenerator().createNewId();
executionPlanContext.getSnapshotService().addSnapshotable(inputHandler.hashCode() + "inputeventhandler", this);
+ mutex = new Semaphore(1);
}
public void sendEvent(Event event) throws InterruptedException {
long eventId = event.getId();
// event id -1 is reserved for the events that are arriving for the first Siddhi node
+ try {
+ mutex.acquire();
if (lastEventId == null || eventId == -1 || lastEventId < eventId) {
lastEventId = eventId;
inputHandler.send(event);
}
+ } finally {
+ mutex.release();
+ }
}
public void sendEvents(Event[] events) throws InterruptedException {
@@ -73,8 +83,14 @@ public class InputEventHandler implements Snapshotable {
@Override
public Map<String, Object> currentState() {
Map<String, Object> state = new HashMap<>();
- synchronized (lastEventId) {
+ try {
+ mutex.acquire();
state.put("LastEventId", lastEventId);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.error("Error when getting the current state", e);
+ } finally {
+ mutex.release();
}
return state;
}
