DefaultIndexRequestFactory implements IndexRequestFactory { @Override public IndexRequest getIndexRequest(OutgoingMessageEnvelope envelope) { IndexRequest indexRequest = getRequest(envelope); getId(envelope).ifPresent(indexRequest::id); getRoutingKey(envelope).ifPresent(indexRequest::routing); getVersion(envelope).ifPresent(indexRequest::version); getVersionType(envelope).ifPresent(indexRequest::versionType); setSource(envelope, indexRequest); return indexRequest; }  @Override IndexRequest getIndexRequest(OutgoingMessageEnvelope envelope);  }
@Test public void testGetIndexRequestStreamName() { IndexRequest indexRequest = INDEX_REQUEST_FACTORY. getIndexRequest(new OutgoingMessageEnvelope(SYSTEM, EMPTY_MSG)); assertEquals(INDEX, indexRequest.index()); assertEquals(TYPE, indexRequest.type()); }
@Test(expected = SamzaException.class) public void testGetIndexRequestInvalidStreamName() { when(SYSTEM.getStream()).thenReturn(INDEX); INDEX_REQUEST_FACTORY.getIndexRequest(new OutgoingMessageEnvelope(SYSTEM, EMPTY_MSG)); }
@Test public void testGetIndexRequestNoId() throws Exception { IndexRequest indexRequest = INDEX_REQUEST_FACTORY.getIndexRequest(new OutgoingMessageEnvelope(SYSTEM, EMPTY_MSG)); assertNull(indexRequest.id()); }
@Test public void testGetIndexRequestWithId() throws Exception { IndexRequest indexRequest = INDEX_REQUEST_FACTORY.getIndexRequest(new OutgoingMessageEnvelope(SYSTEM, "id", EMPTY_MSG)); assertEquals("id", indexRequest.id()); }
@Test public void testGetIndexRequestNoPartitionKey() throws Exception { IndexRequest indexRequest = INDEX_REQUEST_FACTORY.getIndexRequest( new OutgoingMessageEnvelope(SYSTEM, EMPTY_MSG)); assertNull(indexRequest.routing()); }
@Test public void testGetIndexRequestWithPartitionKey() throws Exception { IndexRequest indexRequest = INDEX_REQUEST_FACTORY.getIndexRequest( new OutgoingMessageEnvelope(SYSTEM, "shardKey", "id", EMPTY_MSG)); assertEquals("shardKey", indexRequest.routing()); }
@Test public void testGetIndexRequestMessageBytes() throws Exception { IndexRequest indexRequest = INDEX_REQUEST_FACTORY.getIndexRequest( new OutgoingMessageEnvelope(SYSTEM, "{\"foo\":\"bar\"}".getBytes(Charsets.UTF_8))); assertEquals(Collections.singletonMap("foo", "bar"), indexRequest.sourceAsMap()); }
@Test public void testGetIndexRequestMessageMap() throws Exception { IndexRequest indexRequest = INDEX_REQUEST_FACTORY.getIndexRequest( new OutgoingMessageEnvelope(SYSTEM, Collections.singletonMap("foo", "bar"))); assertEquals(Collections.singletonMap("foo", "bar"), indexRequest.sourceAsMap()); }
@Test(expected = SamzaException.class) public void testGetIndexRequestInvalidMessage() throws Exception { INDEX_REQUEST_FACTORY.getIndexRequest(new OutgoingMessageEnvelope(SYSTEM, "{'foo':'bar'}")); }