PubSubWebSocketAppDataQuery extends PubSubWebSocketInputOperator<String> implements AppData.ConnectionInfoProvider, EmbeddableQueryInfoProvider<String> { public static URI uriHelper(OperatorContext context, URI uri) { if (uri == null) { if (context.getValue(DAG.GATEWAY_CONNECT_ADDRESS) == null) { throw new IllegalArgumentException("The uri property is not set and the dt.attr.GATEWAY_CONNECT_ADDRESS is not defined"); } try { uri = PubSubHelper.getURI(context); } catch (Exception ex) { throw new RuntimeException(ex); } } return uri; } PubSubWebSocketAppDataQuery(); @Override void setup(OperatorContext context); @Override void beginWindow(long windowId); @Override void endWindow(); @Override void teardown(); static URI uriHelper(OperatorContext context, URI uri); @Override URI getUri(); @Override void setUri(URI uri); @Override String getAppDataURL(); @Override DefaultOutputPort<String> getOutputPort(); @Override void enableEmbeddedMode(); long getExecuteIntervalMillis(); void setExecuteIntervalMillis(long executeIntervalMillis); static final long DEFAULT_EXECUTE_INTERVAL_MILLIS; }
@Test public void testURISet() throws Exception { Assert.assertEquals(URI_ADDRESS, PubSubWebSocketAppDataQuery.uriHelper(emptyContext, URI_ADDRESS)); }
@Test public void testNoURISet() throws Exception { boolean threwException = false; try { PubSubWebSocketAppDataQuery.uriHelper(emptyContext, null); } catch (Exception e) { threwException = e instanceof IllegalArgumentException; } Assert.assertTrue(threwException); }
@Test public void testAttrSet() throws Exception { Assert.assertEquals(GATEWAY_CONNECT_ADDRESS, PubSubWebSocketAppDataQuery.uriHelper(context, null)); }
@Test public void testAttrAndURISet() throws Exception { Assert.assertEquals(URI_ADDRESS, PubSubWebSocketAppDataQuery.uriHelper(context, URI_ADDRESS)); }