@Test(timeout = 30000L) public void testPutAll() throws Exception { List<Event> events = Lists.newArrayList(); events.add(EventBuilder.withBody(body, headers)); agent.configure(properties); agent.start(); agent.putAll(events); Event event; while((event = eventCollector.poll()) == null) { Thread.sleep(500L); } Assert.assertNotNull(event); Assert.assertArrayEquals(body, event.getBody()); Assert.assertEquals(headers, event.getHeaders()); }
public void putAll(List<Event> events) throws EventDeliveryException { if(state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.putAll(events); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } }
EmbeddedAgent { public void putAll(List<Event> events) throws EventDeliveryException { if(state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.putAll(events); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } } }
EmbeddedAgent { public void putAll(List<Event> events) throws EventDeliveryException { if(state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.putAll(events); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } } @InterfaceAudience.Private @InterfaceStability.Unstable @VisibleForTesting  EmbeddedAgent(MaterializedConfigurationProvider configurationProvider,
      String name);  EmbeddedAgent(String name); }
EmbeddedAgent { public void putAll(List<Event> events) throws EventDeliveryException { if(state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.putAll(events); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } } @InterfaceAudience.Private @InterfaceStability.Unstable @VisibleForTesting  EmbeddedAgent(MaterializedConfigurationProvider configurationProvider,
      String name);  EmbeddedAgent(String name); void configure(Map<String, String> properties); void start(); void stop(); void put(Event event); void putAll(List<Event> events); }
EmbeddedAgent { public void putAll(List<Event> events) throws EventDeliveryException { if(state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.putAll(events); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } } @InterfaceAudience.Private @InterfaceStability.Unstable @VisibleForTesting  EmbeddedAgent(MaterializedConfigurationProvider configurationProvider,
      String name);  EmbeddedAgent(String name); void configure(Map<String, String> properties); void start(); void stop(); void put(Event event); void putAll(List<Event> events);  }