HttpSink extends AbstractSink implements Configurable { @Override public final void configure(final Context context) { String configuredEndpoint = context.getString("endpoint", ""); LOG.info("Read endpoint URL from configuration : " + configuredEndpoint); try { endpointUrl = new URL(configuredEndpoint); } catch (MalformedURLException e) { throw new IllegalArgumentException("Endpoint URL invalid", e); } connectTimeout = context.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); if (connectTimeout <= 0) { throw new IllegalArgumentException( "Connect timeout must be a non-zero and positive"); } LOG.info("Using connect timeout : " + connectTimeout); requestTimeout = context.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); if (requestTimeout <= 0) { throw new IllegalArgumentException( "Request timeout must be a non-zero and positive"); } LOG.info("Using request timeout : " + requestTimeout); acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER); LOG.info("Using Accept header value : " + acceptHeader); contentTypeHeader = context.getString("contentTypeHeader", DEFAULT_CONTENT_TYPE); LOG.info("Using Content-Type header value : " + contentTypeHeader); defaultBackoff = context.getBoolean("defaultBackoff", true); LOG.info("Channel backoff by default is " + defaultBackoff); defaultRollback = context.getBoolean("defaultRollback", true); LOG.info("Transaction rollback by default is " + defaultRollback); defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics", false); LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics); parseConfigOverrides("backoff", context, backoffOverrides); parseConfigOverrides("rollback", context, rollbackOverrides); parseConfigOverrides("incrementMetrics", context, incrementMetricsOverrides); if (this.sinkCounter == null) { this.sinkCounter = new SinkCounter(this.getName()); } connectionBuilder = new ConnectionBuilder(); }  @Override final void configure(final Context context); @Override final void start(); @Override final void stop(); @Override final Status process();  }
@Test public void ensureAllConfigurationOptionsRead() { whenDefaultStringConfig(); whenDefaultBooleanConfig(); when(configContext.getInteger(eq("connectTimeout"), Mockito.anyInt())).thenReturn(1000); when(configContext.getInteger(eq("requestTimeout"), Mockito.anyInt())).thenReturn(1000); new HttpSink().configure(configContext); verify(configContext).getString("endpoint", ""); verify(configContext).getInteger(eq("connectTimeout"), Mockito.anyInt()); verify(configContext).getInteger(eq("requestTimeout"), Mockito.anyInt()); verify(configContext).getString(eq("acceptHeader"), Mockito.anyString()); verify(configContext).getString(eq("contentTypeHeader"), Mockito.anyString()); verify(configContext).getBoolean("defaultBackoff", true); verify(configContext).getBoolean("defaultRollback", true); verify(configContext).getBoolean("defaultIncrementMetrics", false); }
@Test(expected = IllegalArgumentException.class) public void ensureExceptionIfEndpointUrlEmpty() { when(configContext.getString("endpoint", "")).thenReturn(""); new HttpSink().configure(configContext); }
@Test(expected = IllegalArgumentException.class) public void ensureExceptionIfEndpointUrlInvalid() { when(configContext.getString("endpoint", "")).thenReturn("invalid url"); new HttpSink().configure(configContext); }
@Test(expected = IllegalArgumentException.class) public void ensureExceptionIfConnectTimeoutNegative() { whenDefaultStringConfig(); when(configContext.getInteger("connectTimeout", 1000)).thenReturn(-1000); when(configContext.getInteger(eq("requestTimeout"), Mockito.anyInt())).thenReturn(1000); new HttpSink().configure(configContext); }
@Test public void ensureDefaultConnectTimeoutCorrect() { whenDefaultStringConfig(); when(configContext.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT)).thenReturn(1000); when(configContext.getInteger(eq("requestTimeout"), Mockito.anyInt())).thenReturn(1000); new HttpSink().configure(configContext); verify(configContext).getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); }
@Test(expected = IllegalArgumentException.class) public void ensureExceptionIfRequestTimeoutNegative() { whenDefaultStringConfig(); when(configContext.getInteger("requestTimeout", 1000)).thenReturn(-1000); when(configContext.getInteger(eq("connectTimeout"), Mockito.anyInt())).thenReturn(1000); new HttpSink().configure(configContext); }
@Test public void ensureDefaultRequestTimeoutCorrect() { whenDefaultStringConfig(); when(configContext.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT)).thenReturn(1000); when(configContext.getInteger(eq("connectTimeout"), Mockito.anyInt())).thenReturn(1000); new HttpSink().configure(configContext); verify(configContext).getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); }
@Test public void ensureDefaultAcceptHeaderCorrect() { whenDefaultTimeouts(); whenDefaultStringConfig(); new HttpSink().configure(configContext); verify(configContext).getString("acceptHeader", DEFAULT_ACCEPT_HEADER); }
@Test public void ensureDefaultContentTypeHeaderCorrect() { whenDefaultTimeouts(); whenDefaultStringConfig(); new HttpSink().configure(configContext); verify(configContext).getString("contentTypeHeader", DEFAULT_CONTENT_TYPE_HEADER); }