CEPSourceFunction.java

public class DummySource implements SourceFunction<String> {

	private static final long serialVersionUID = 3978123556403297086L;
//	private static Queue<String> queue = new LinkedBlockingQueue<String>();
	private Queue<String> queue;
	private boolean cancel = false;

	public void setQueue(Queue<String> q){
		queue = q;
	}	

	@Override
	public void run(org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext<String> ctx)
			throws Exception {
		System.out.println("run");
		synchronized (queue) {			
			while (!cancel) {
				if (queue.peek() != null) {
					String e = queue.poll();
					if (e.equals("exit")) {
						cancel();
					}
					System.out.println("collect "+e);
					ctx.collectWithTimestamp(e, System.currentTimeMillis());
				}
			}
		}
	}

	@Override
	public void cancel() {
		System.out.println("canceled");
		cancel = true;
	}
}