CODEWITHSUNDEEP
javaesper
AHH
AHH

Reputation: 1083

Communication of separate processes through Esper events

I am trying to let multiple java processes exchange events using Esper. One process should send events, the other prepares a query and reacts according to the reported events.

When both operations are done within the same java process, everything works fine. But when I use two different processes, they just don't see each other. I am wondering what is the key for this communication. I used the same name for the provider. This is all I could do so far.

The Producer:

String aType = espertest.dummy.A.class.getName();
Configuration cepConfig = new Configuration();
cepConfig.addEventType("A",aType);
EPServiceProvider epService = EPServiceProviderManager.getProvider("DummyProvider", cepConfig);
Object o = new A();
epService.getEPRuntime().sendEvent(o);

The Consumer:

String aType = A.class.getName();
String expression = "select count(*) from "+aType + "";
System.out.println("Our Query: " + expression);

Configuration cepConfig = new Configuration();
cepConfig.addEventType("A",aType);

EPServiceProvider epService = EPServiceProviderManager.getProvider("DummyProvider", cepConfig);

EPStatement statement = epService.getEPAdministrator().createEPL(expression);

DummyListener listener = new DummyListener();
statement.addListener(listener);
System.out.println("Anything");


try{
    A a = new A();
    epService.getEPRuntime().sendEvent(a);
    Thread.sleep(60000);
}catch(Exception E)
{
    System.out.println("Exception ");
}

The consumer tries to count the events of type A. It also sends an instance of A as a test, and this works fine. The listener is called as expected. The code above is just an excerpt.

Upvotes: 1

Views: 348

Answers (1)

R Dub
R Dub

Reputation: 688

You need to configure middleware (Message Queue, Distributed Cache, Networked FileSystem, Socket Connection, etc....) to get the events from the producer JVM to the consumer JVM. If you can deploy the producer and consumer to a container that supports Apache Camel (e.g. ServiceMix) then it should be trivial to stand up a prototype that uses ActiveMQ to transport your objects into Esper as Camel has support for both products.

JVM 1

  • From Data Source
  • To CEP Engine 1
  • To Message Queue

JVM 2 (also could host MQ Broker)

  • From Message Queue
  • To CEP Engine 2
  • To Destination

Update:

If the producer and consumer can be threads in the same JVM, then the issue may be in the consumer. I cannot see where the consumer does anything with the event from the producer. Try something like this instead (esper reference is provided to the producer/consumer and consumer is reworked with an update method to handle results of the select statement).

Test Driver:

public Driver() {
    String aType = espertest.dummy.A.class.getName();
    Configuration cepConfig = new Configuration();
    cepConfig.addEventType("A",aType);
    EPServiceProvider epService = EPServiceProviderManager.getProvider("DummyProvider", cepConfig);
    Consumer c = new Consumer(epService);
    Producer p = new Producer(epService);
}

Producer:

public Producer(EPServiceProvider epsp) {
    Object o = new A();
    epsp.getEPRuntime().sendEvent(o);
}

Consumer:

public Consumer(EPServiceProvider epsp) {
    EPStatement statement = epsp.getEPAdministrator().createEPL(input);
    statement.setSubscriber(this);
}
public void update(A event) {
    System.out.println("Consumer received event!");    
}

Upvotes: 1

Related Questions