@Test(timeout = 30000L) public void testPut() throws Exception { agent.configure(properties); agent.start(); agent.put(EventBuilder.withBody(body, headers)); 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 put(Event event) throws EventDeliveryException { if (state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.put(event); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } }
EmbeddedAgent { public void put(Event event) throws EventDeliveryException { if (state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.put(event); } catch (ChannelException ex) { throw new EventDeliveryException("Embedded agent " + name + ": Unable to process event: " + ex.getMessage(), ex); } } }
EmbeddedAgent { public void put(Event event) throws EventDeliveryException { if (state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.put(event); } 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 put(Event event) throws EventDeliveryException { if (state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.put(event); } 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 put(Event event) throws EventDeliveryException { if (state != State.STARTED) { throw new IllegalStateException("Cannot put events unless started"); } try { embeddedSource.put(event); } 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);  }