NakadiClient { public <T> void publish(String eventName, List<T> events) throws IOException { final URI uri = baseUri.resolve(String.format("/event-types/%s/events", eventName)); final Request request = clientHttpRequestFactory.createRequest(uri, "POST"); request.getHeaders().setContentType(ContentType.APPLICATION_JSON); try (final OutputStream body = request.getBody()) { objectMapper.writeValue(body, events); } try (final Response response = request.execute()) { LOG.debug("Successfully published [{}] events for [{}]", events.size(), eventName); } } NakadiClient(URI baseUri, RequestFactory clientHttpRequestFactory, ObjectMapper objectMapper, CursorManager cursorManager); static NakadiClientBuilder builder(URI baseUri); List<Partition> getPartitions(String eventName); void publish(String eventName, List<T> events); @Deprecated Subscription subscribe(String applicationName, String eventName, String consumerGroup); SubscriptionBuilder subscription(String applicationName, String eventName); SubscriptionBuilder subscription(String applicationName, Set<String> eventNames); void deleteSubscription(String subscriptionId); StreamBuilder.SubscriptionStreamBuilder stream(Subscription subscription); StreamBuilder.LowLevelStreamBuilder stream(String eventName);  }
@Test public void shouldPublishEvents() throws IOException { server.expectRequestTo("http: .andExpectJsonPath("$[0].id", equalTo("1")) .andExpectJsonPath("$[1].id", equalTo("2")) .andRespondWith(200) .setup(); client.publish("foobar", asList(new SomeEvent("1"), new SomeEvent("2"))); }
@Test public void shouldHandleMultiStatusWhenPublishing() throws IOException { server.expectRequestTo("http: .andExpectJsonPath("$[0].id", equalTo("1")) .andExpectJsonPath("$[1].id", equalTo("2")) .andRespondWith(207, ContentType.APPLICATION_JSON, "[{\"publishing_status\":\"failed\",\"step\":\"validating\",\"detail\":\"baz\"}]") .setup(); expectedException.expect(EventPublishingException.class); expectedException.expectMessage("returned status [failed] in step [validating] with detail [baz]"); client.publish("foobar", asList(new SomeEvent("1"), new SomeEvent("2"))); server.verify(); }
@Test public void shouldHandleSuccessFulMultiStatusWhenPublishing() throws IOException { server.expectRequestTo("http: .andExpectJsonPath("$[0].id", equalTo("1")) .andExpectJsonPath("$[1].id", equalTo("2")) .andRespondWith(207, ContentType.APPLICATION_JSON, "[{\"publishing_status\":\"submitted\"}, {\"publishing_status\":\"submitted\"}]") .setup(); client.publish("foobar", asList(new SomeEvent("1"), new SomeEvent("2"))); server.verify(); }
@Test public void shouldHandleBatchItemResponseWhenPublishing() throws IOException { server.expectRequestTo("http: .andExpectJsonPath("$[0].id", equalTo("1")) .andExpectJsonPath("$[1].id", equalTo("2")) .andRespondWith(422, ContentType.APPLICATION_JSON, "[{\"publishing_status\":\"aborted\",\"step\":\"publishing\",\"detail\":\"baz\"}]") .setup(); expectedException.expect(EventPublishingException.class); expectedException.expectMessage("returned status [aborted] in step [publishing] with detail [baz]"); client.publish("foobar", asList(new SomeEvent("1"), new SomeEvent("2"))); server.verify(); }