### Question:
StreamSubscriber implements Subscriber { @Override public SubscriptionContext subscribeToStream(URL wpsUrl, String stream, DataFilter filter, Geometry g) { LOGGER.info("Performing request " + StreamingConstants.SUBSCRIBE_PROCESS + " - " + wpsUrl + " for layer " + stream + " using data filter [" + filter + "] and spatial filter [" + g + "]"); InputStream response = getResponseViaPost(wpsUrl, buildRequest(stream, filter, normalize(g))); if (response == null) { return null; } try { Configuration cfg = XMLUtilities.readXMLObject(response, Configuration.class); if (cfg == null || cfg.getUrl() == null) { return null; } String filterId = cfg.getFilterId(); String idParam = cfg.getUrl().getRequestParameters().get(0); StringBuilder buf = new StringBuilder(); buf.append(cfg.getUrl().getBase()); buf.append('?'); buf.append(idParam); buf.append('='); buf.append(filterId); buf.append('&'); buf.append(cfg.getUrl().getRequestParameters().get(1)); buf.append('='); int pollInterval = Integer.parseInt(System.getProperty("NRTPollInterval", "5000")); buf.append(pollInterval); URL url = new URL(buf.toString()); LOGGER.info("Stream URL for layer " + stream + " is " + url); SubscriptionContext context = new SubscriptionContext(); context.setFilterIdParameterName(idParam); context.setStreamId(UUID.fromString(filterId)); context.setStreamUrl(url); context.setPollInterval(pollInterval); return context; } catch (JAXBException | MalformedURLException e) { LOGGER.error(e.getMessage(), e); } return null; } StreamSubscriber(ServerProviderRegistry providerRegistry); ServerProviderRegistry getServerProviderRegistry(); @Override SubscriptionContext subscribeToStream(URL wpsUrl, String stream, DataFilter filter, Geometry g);  }### Answer:
@Test public void testSubscribeToStream() throws IOException, URISyntaxException { EasyMockSupport support = new EasyMockSupport(); ServerProviderRegistry registry = createRegistry(support, false, false); support.replayAll(); StreamSubscriber subscriber = new StreamSubscriber(registry); SubscriptionContext context = subscriber.subscribeToStream(new URL(ourWpsUrl), ourLayerName, null, null); assertEquals("29d25e04-6884-4ed5-88e9-1a93773a6743", context.getStreamId().toString()); assertEquals("https: context.getStreamUrl().toString()); assertEquals("filterId", context.getFilterIdParameterName()); assertEquals(5000, context.getPollInterval()); support.verifyAll(); }
@Test public void testSubscribeToStreamBadResponse() throws IOException, URISyntaxException { EasyMockSupport support = new EasyMockSupport(); ServerProviderRegistry registry = createRegistry(support, true, false); support.replayAll(); StreamSubscriber subscriber = new StreamSubscriber(registry); SubscriptionContext context = subscriber.subscribeToStream(new URL(ourWpsUrl), ourLayerName, null, null); assertNull(context); support.verifyAll(); }
@Test public void testSubscribeToStreamWithFilter() throws IOException, URISyntaxException { EasyMockSupport support = new EasyMockSupport(); ServerProviderRegistry registry = createRegistry(support, false, true); support.replayAll(); ImmutableDataFilterCriteria criteria = new ImmutableDataFilterCriteria(ourColumnName, ourColumnValue, Conditional.EQ, null); ImmutableDataFilterGroup group = new ImmutableDataFilterGroup("group", Logical.AND, New.list(criteria), New.<DataFilterGroup>list(), null); ImmutableDataFilter filter = new ImmutableDataFilter("filter", "type", New.list("column1"), group, null, null); StreamSubscriber subscriber = new StreamSubscriber(registry); SubscriptionContext context = subscriber.subscribeToStream(new URL(ourWpsUrl), ourLayerName, filter, null); assertEquals("29d25e04-6884-4ed5-88e9-1a93773a6743", context.getStreamId().toString()); assertEquals("https: context.getStreamUrl().toString()); assertEquals("filterId", context.getFilterIdParameterName()); support.verifyAll(); }