EmbeddedEPNManager extends AbstractEPNManager { public void publish(String subject, java.util.List<? extends java.io.Serializable> events) throws Exception { publish(subject, new EventList(events)); }  void publish(String subject, java.util.List<? extends java.io.Serializable> events); void close();  }
@Test public void testEventTransformedForChild() { Network net=new Network(); net.setName(TEST_NETWORK); net.setVersion(VER1); Subscription sub=new Subscription(); sub.setSubject(TEST_SUBJECT); sub.setNodeName(N1); net.getSubscriptions().add(sub); Node n1=new Node(); n1.setName(N1); n1.setPredicate(new TestPredicate1()); n1.setEventProcessor(new TestEventProcessorB()); net.getNodes().add(n1); Node n2=new Node(); n2.setPredicate(new TestPredicate2()); TestEventProcessorA tea=new TestEventProcessorA(); n2.setName(N2); n2.setEventProcessor(tea); n2.getSourceNodes().add(N1); net.getNodes().add( n2); EmbeddedEPNManager mgr=new EmbeddedEPNManager(); try { mgr.register(net); java.util.List<Serializable> eventsList=new java.util.ArrayList<Serializable>(); EventList events=new EventList(eventsList); TestEvent1 te1=new TestEvent1(25); eventsList.add(te1); mgr.publish(TEST_SUBJECT, events); Thread.sleep(1000); if (tea.getEvents().size() != 1) { fail("Expecting 1 event to be processed: "+tea.getEvents().size()); } if (tea.getEvents().get(0) instanceof TestEvent2) { TestEvent2 te2=(TestEvent2)tea.getEvents().get(0); if (te2.getValue() != 25) { fail("Value is not correct: "+te2.getValue()); } } else { fail("Incorrect event type: "+tea.getEvents().get(0).getClass()); } } catch(Exception e) { e.printStackTrace(); fail("Test failed: "+e); } }
@Test public void testEventRetries() { Network net=new Network(); net.setName(TEST_NETWORK); net.setVersion(VER1); Subscription sub=new Subscription(); sub.setSubject(TEST_SUBJECT); sub.setNodeName(N1); net.getSubscriptions().add(sub); Node n1=new Node(); n1.setName(N1); n1.setPredicate(new TestPredicate1()); n1.setEventProcessor(new TestEventProcessorB()); net.getNodes().add(n1); Node n2=new Node(); n2.setName(N2); n2.setPredicate(new TestPredicate2()); n2.setMaxRetries(MAX_RETRIES); n2.getSourceNodes().add(N1); TestEventProcessorC tea=new TestEventProcessorC(); n2.setEventProcessor(tea); net.getNodes().add(n2); EmbeddedEPNManager mgr=new EmbeddedEPNManager(); try { mgr.register(net); java.util.List<Serializable> eventsList=new java.util.ArrayList<Serializable>(); EventList events=new EventList(eventsList); TestEvent1 te1=new TestEvent1(25); eventsList.add(te1); mgr.publish(TEST_SUBJECT, events); Thread.sleep(1000); if (tea._retryCount != MAX_RETRIES+1) { fail("Expecting "+(MAX_RETRIES+1)+" retries: "+tea._retryCount); } } catch(Exception e) { fail("Test failed: "+e); } }