CouchbaseProducer extends DefaultProducer { @Override public void process(Exchange exchange) throws Exception { Map<String, Object> headers = exchange.getIn().getHeaders(); String id = (headers.containsKey(HEADER_ID)) ? exchange.getIn().getHeader(HEADER_ID, String.class) : endpoint.getId(); int ttl = (headers.containsKey(HEADER_TTL)) ? Integer.parseInt(exchange.getIn().getHeader(HEADER_TTL, String.class)) : DEFAULT_TTL; if (endpoint.isAutoStartIdForInserts()) { id = Long.toString(startId); startId++; } else if (id == null) { throw new CouchbaseException(HEADER_ID + " is not specified in message header or endpoint URL.", exchange); } if (endpoint.getOperation().equals(COUCHBASE_PUT)) { log.info("Type of operation: PUT"); Object obj = exchange.getIn().getBody(); exchange.getOut().setBody(setDocument(id, ttl, obj, persistTo, replicateTo)); } else if (endpoint.getOperation().equals(COUCHBASE_GET)) { log.info("Type of operation: GET"); Object result = client.get(id); exchange.getOut().setBody(result); } else if (endpoint.getOperation().equals(COUCHBASE_DELETE)) { log.info("Type of operation: DELETE"); Future<Boolean> result = client.delete(id); exchange.getOut().setBody(result.get()); } exchange.getIn().removeHeader(HEADER_ID); } CouchbaseProducer(CouchbaseEndpoint endpoint, CouchbaseClientIF client, int persistTo, int replicateTo); @Override void process(Exchange exchange);  }
@SuppressWarnings("unchecked") @Test(expected = CouchbaseException.class) public void testBodyMandatory() throws Exception { when(msg.getMandatoryBody()).thenThrow(InvalidPayloadException.class); producer.process(exchange); }
@Test public void testExpiryTimeIsSet() throws Exception { OperationFuture of = mock(OperationFuture.class); when(of.get()).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { return true; } }); when( client.set(org.mockito.Matchers.anyString(), org.mockito.Matchers.anyInt(), org.mockito.Matchers.anyObject(), org.mockito.Matchers.any(PersistTo.class), org.mockito.Matchers.any(ReplicateTo.class))).thenReturn(of); int expiry = 5000; Map<String, Object> testHeaders = new HashMap<String, Object>(); testHeaders.put("CCB_TTL", Integer.toString(expiry)); when(msg.getHeaders()).thenReturn(testHeaders); when(msg.getHeader(HEADER_TTL, String.class)).thenReturn(Integer.toString(expiry)); when(endpoint.getId()).thenReturn("123"); when(endpoint.getOperation()).thenReturn("CCB_PUT"); Message outmsg = mock(Message.class); when(exchange.getOut()).thenReturn(msg); producer.process(exchange); verify(client).set(org.mockito.Matchers.anyString(), Mockito.eq(expiry), org.mockito.Matchers.anyObject(), org.mockito.Matchers.any(PersistTo.class), org.mockito.Matchers.any(ReplicateTo.class)); }
@Test public void testTimeOutRetryToException() throws Exception { OperationFuture of = mock(OperationFuture.class); when(of.get()).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { throw new RuntimeException("Timed out waiting for operation"); } }); when( client.set(org.mockito.Matchers.anyString(), org.mockito.Matchers.anyInt(), org.mockito.Matchers.anyObject(), org.mockito.Matchers.any(PersistTo.class), org.mockito.Matchers.any(ReplicateTo.class))).thenReturn(of); when(endpoint.getId()).thenReturn("123"); when(endpoint.getOperation()).thenReturn("CCB_PUT"); try { producer.process(exchange); } catch (Exception e) { verify(of, times(3)).get(); } }
@Test public void testTimeOutRetryThenSuccess() throws Exception { OperationFuture of = mock(OperationFuture.class); when(of.get()).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { throw new RuntimeException("Timed out waiting for operation"); } }).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { return true; } }); when( client.set(org.mockito.Matchers.anyString(), org.mockito.Matchers.anyInt(), org.mockito.Matchers.anyObject(), org.mockito.Matchers.any(PersistTo.class), org.mockito.Matchers.any(ReplicateTo.class))).thenReturn(of); when(endpoint.getId()).thenReturn("123"); when(endpoint.getOperation()).thenReturn("CCB_PUT"); when(exchange.getOut()).thenReturn(msg); producer.process(exchange); verify(of, times(2)).get(); verify(msg).setBody(true); }
@Test public void testTimeOutRetryTwiceThenSuccess() throws Exception { OperationFuture of = mock(OperationFuture.class); when(of.get()).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { throw new RuntimeException("Timed out waiting for operation"); } }).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { throw new RuntimeException("Timed out waiting for operation"); } }).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Exception { return true; } }); when( client.set(org.mockito.Matchers.anyString(), org.mockito.Matchers.anyInt(), org.mockito.Matchers.anyObject(), org.mockito.Matchers.any(PersistTo.class), org.mockito.Matchers.any(ReplicateTo.class))).thenReturn(of); when(endpoint.getId()).thenReturn("123"); when(endpoint.getOperation()).thenReturn("CCB_PUT"); when(exchange.getOut()).thenReturn(msg); producer.process(exchange); verify(of, times(3)).get(); verify(msg).setBody(true); }