@Test public void streamChunkedResponseTest() throws IOException { String channelCode = "mock_collection"; String callbackName = null; Float rate = new Float(-1); String duration = "-1"; ChunkedOutput<String> output = asyncStream.streamChunkedResponse(channelCode, callbackName, rate, duration); Class<?> s = output.getRawType(); assertNotNull(output); assertEquals("class java.lang.String", s.toString()); }
@GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) public ChunkedOutput<String> streamChunkedResponse( @PathParam("crisisCode") String channelCode, @QueryParam("callback") final String callbackName, @DefaultValue("-1") @QueryParam("rate") Float rate, @DefaultValue("-1") @QueryParam("duration") String duration) throws IOException { final ChunkedOutput<String> responseWriter = new ChunkedOutput<String>(String.class); writerList.add(responseWriter); logger.info("Received streaming request for channelCode=" + channelCode + ", rate=" + rate + ", duration=" + duration + ", callbackName=" + callbackName); if (channelCode != null) { if (jedisConn != null) { Jedis subscriberJedis = jedisConn.getJedisResource(); String channel = setFullyQualifiedChannelName(CHANNEL_PREFIX_CODE, channelCode); SubscriptionDataObject subscriptionData = new SubscriptionDataObject(); subscriptionData.rejectNullFlag = rejectNullFlag; subscriptionData.jedisConn = jedisConn; subscriptionData.subscriberJedis = subscriberJedis; subscriptionData.redisChannel = channel; subscriptionData.patternSubscriptionFlag = isPattern(channelCode); subscriptionData.callbackName = callbackName; subscriptionData.rate = rate; subscriptionData.duration = duration.equals("-1") ? null : duration; AsyncStreamRedisSubscriber aidrSubscriber = new AsyncStreamRedisSubscriber(subscriberJedis, responseWriter, writerList, subscriptionData); try { subscribeToChannel(executorService, aidrSubscriber, subscriptionData); jedisConn.setJedisSubscription(subscriberJedis, subscriptionData.patternSubscriptionFlag); } catch (Exception e) { logger.error(channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); OutputErrorHandler.sendErrorMail(e.getLocalizedMessage(), channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); } logger.info(channelCode + ": Spawning async response thread"); try { executorService.execute(aidrSubscriber); } catch (RejectedExecutionException|NullPointerException e) { logger.error(channelCode + "Fatal error executing async thread! Terminating."); } } } else { JsonObject errorMessage = new JsonObject(); errorMessage.addProperty("crisisCode", "NOT PROVIDED"); errorMessage.addProperty("streaming status", "ERROR"); if (callbackName != null) { responseWriter.write(callbackName + "(" + errorMessage.getAsString() + ")"); } else { responseWriter.write(errorMessage.getAsString()); } } return responseWriter; }
AsyncStream implements ServletContextListener { @GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) public ChunkedOutput<String> streamChunkedResponse( @PathParam("crisisCode") String channelCode, @QueryParam("callback") final String callbackName, @DefaultValue("-1") @QueryParam("rate") Float rate, @DefaultValue("-1") @QueryParam("duration") String duration) throws IOException { final ChunkedOutput<String> responseWriter = new ChunkedOutput<String>(String.class); writerList.add(responseWriter); logger.info("Received streaming request for channelCode=" + channelCode + ", rate=" + rate + ", duration=" + duration + ", callbackName=" + callbackName); if (channelCode != null) { if (jedisConn != null) { Jedis subscriberJedis = jedisConn.getJedisResource(); String channel = setFullyQualifiedChannelName(CHANNEL_PREFIX_CODE, channelCode); SubscriptionDataObject subscriptionData = new SubscriptionDataObject(); subscriptionData.rejectNullFlag = rejectNullFlag; subscriptionData.jedisConn = jedisConn; subscriptionData.subscriberJedis = subscriberJedis; subscriptionData.redisChannel = channel; subscriptionData.patternSubscriptionFlag = isPattern(channelCode); subscriptionData.callbackName = callbackName; subscriptionData.rate = rate; subscriptionData.duration = duration.equals("-1") ? null : duration; AsyncStreamRedisSubscriber aidrSubscriber = new AsyncStreamRedisSubscriber(subscriberJedis, responseWriter, writerList, subscriptionData); try { subscribeToChannel(executorService, aidrSubscriber, subscriptionData); jedisConn.setJedisSubscription(subscriberJedis, subscriptionData.patternSubscriptionFlag); } catch (Exception e) { logger.error(channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); OutputErrorHandler.sendErrorMail(e.getLocalizedMessage(), channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); } logger.info(channelCode + ": Spawning async response thread"); try { executorService.execute(aidrSubscriber); } catch (RejectedExecutionException|NullPointerException e) { logger.error(channelCode + "Fatal error executing async thread! Terminating."); } } } else { JsonObject errorMessage = new JsonObject(); errorMessage.addProperty("crisisCode", "NOT PROVIDED"); errorMessage.addProperty("streaming status", "ERROR"); if (callbackName != null) { responseWriter.write(callbackName + "(" + errorMessage.getAsString() + ")"); } else { responseWriter.write(errorMessage.getAsString()); } } return responseWriter; } }
AsyncStream implements ServletContextListener { @GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) public ChunkedOutput<String> streamChunkedResponse( @PathParam("crisisCode") String channelCode, @QueryParam("callback") final String callbackName, @DefaultValue("-1") @QueryParam("rate") Float rate, @DefaultValue("-1") @QueryParam("duration") String duration) throws IOException { final ChunkedOutput<String> responseWriter = new ChunkedOutput<String>(String.class); writerList.add(responseWriter); logger.info("Received streaming request for channelCode=" + channelCode + ", rate=" + rate + ", duration=" + duration + ", callbackName=" + callbackName); if (channelCode != null) { if (jedisConn != null) { Jedis subscriberJedis = jedisConn.getJedisResource(); String channel = setFullyQualifiedChannelName(CHANNEL_PREFIX_CODE, channelCode); SubscriptionDataObject subscriptionData = new SubscriptionDataObject(); subscriptionData.rejectNullFlag = rejectNullFlag; subscriptionData.jedisConn = jedisConn; subscriptionData.subscriberJedis = subscriberJedis; subscriptionData.redisChannel = channel; subscriptionData.patternSubscriptionFlag = isPattern(channelCode); subscriptionData.callbackName = callbackName; subscriptionData.rate = rate; subscriptionData.duration = duration.equals("-1") ? null : duration; AsyncStreamRedisSubscriber aidrSubscriber = new AsyncStreamRedisSubscriber(subscriberJedis, responseWriter, writerList, subscriptionData); try { subscribeToChannel(executorService, aidrSubscriber, subscriptionData); jedisConn.setJedisSubscription(subscriberJedis, subscriptionData.patternSubscriptionFlag); } catch (Exception e) { logger.error(channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); OutputErrorHandler.sendErrorMail(e.getLocalizedMessage(), channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); } logger.info(channelCode + ": Spawning async response thread"); try { executorService.execute(aidrSubscriber); } catch (RejectedExecutionException|NullPointerException e) { logger.error(channelCode + "Fatal error executing async thread! Terminating."); } } } else { JsonObject errorMessage = new JsonObject(); errorMessage.addProperty("crisisCode", "NOT PROVIDED"); errorMessage.addProperty("streaming status", "ERROR"); if (callbackName != null) { responseWriter.write(callbackName + "(" + errorMessage.getAsString() + ")"); } else { responseWriter.write(errorMessage.getAsString()); } } return responseWriter; }  }
AsyncStream implements ServletContextListener { @GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) public ChunkedOutput<String> streamChunkedResponse( @PathParam("crisisCode") String channelCode, @QueryParam("callback") final String callbackName, @DefaultValue("-1") @QueryParam("rate") Float rate, @DefaultValue("-1") @QueryParam("duration") String duration) throws IOException { final ChunkedOutput<String> responseWriter = new ChunkedOutput<String>(String.class); writerList.add(responseWriter); logger.info("Received streaming request for channelCode=" + channelCode + ", rate=" + rate + ", duration=" + duration + ", callbackName=" + callbackName); if (channelCode != null) { if (jedisConn != null) { Jedis subscriberJedis = jedisConn.getJedisResource(); String channel = setFullyQualifiedChannelName(CHANNEL_PREFIX_CODE, channelCode); SubscriptionDataObject subscriptionData = new SubscriptionDataObject(); subscriptionData.rejectNullFlag = rejectNullFlag; subscriptionData.jedisConn = jedisConn; subscriptionData.subscriberJedis = subscriberJedis; subscriptionData.redisChannel = channel; subscriptionData.patternSubscriptionFlag = isPattern(channelCode); subscriptionData.callbackName = callbackName; subscriptionData.rate = rate; subscriptionData.duration = duration.equals("-1") ? null : duration; AsyncStreamRedisSubscriber aidrSubscriber = new AsyncStreamRedisSubscriber(subscriberJedis, responseWriter, writerList, subscriptionData); try { subscribeToChannel(executorService, aidrSubscriber, subscriptionData); jedisConn.setJedisSubscription(subscriberJedis, subscriptionData.patternSubscriptionFlag); } catch (Exception e) { logger.error(channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); OutputErrorHandler.sendErrorMail(e.getLocalizedMessage(), channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); } logger.info(channelCode + ": Spawning async response thread"); try { executorService.execute(aidrSubscriber); } catch (RejectedExecutionException|NullPointerException e) { logger.error(channelCode + "Fatal error executing async thread! Terminating."); } } } else { JsonObject errorMessage = new JsonObject(); errorMessage.addProperty("crisisCode", "NOT PROVIDED"); errorMessage.addProperty("streaming status", "ERROR"); if (callbackName != null) { responseWriter.write(callbackName + "(" + errorMessage.getAsString() + ")"); } else { responseWriter.write(errorMessage.getAsString()); } } return responseWriter; }  @Override void contextInitialized(ServletContextEvent sce); @Override void contextDestroyed(ServletContextEvent sce); String setFullyQualifiedChannelName(final String channelPrefixCode, final String channelCode); @GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) ChunkedOutput<String> streamChunkedResponse(
			@PathParam("crisisCode") String channelCode,
			@QueryParam("callback") final String callbackName,
			@DefaultValue("-1") @QueryParam("rate") Float rate,
			@DefaultValue("-1") @QueryParam("duration") String duration); }
AsyncStream implements ServletContextListener { @GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) public ChunkedOutput<String> streamChunkedResponse( @PathParam("crisisCode") String channelCode, @QueryParam("callback") final String callbackName, @DefaultValue("-1") @QueryParam("rate") Float rate, @DefaultValue("-1") @QueryParam("duration") String duration) throws IOException { final ChunkedOutput<String> responseWriter = new ChunkedOutput<String>(String.class); writerList.add(responseWriter); logger.info("Received streaming request for channelCode=" + channelCode + ", rate=" + rate + ", duration=" + duration + ", callbackName=" + callbackName); if (channelCode != null) { if (jedisConn != null) { Jedis subscriberJedis = jedisConn.getJedisResource(); String channel = setFullyQualifiedChannelName(CHANNEL_PREFIX_CODE, channelCode); SubscriptionDataObject subscriptionData = new SubscriptionDataObject(); subscriptionData.rejectNullFlag = rejectNullFlag; subscriptionData.jedisConn = jedisConn; subscriptionData.subscriberJedis = subscriberJedis; subscriptionData.redisChannel = channel; subscriptionData.patternSubscriptionFlag = isPattern(channelCode); subscriptionData.callbackName = callbackName; subscriptionData.rate = rate; subscriptionData.duration = duration.equals("-1") ? null : duration; AsyncStreamRedisSubscriber aidrSubscriber = new AsyncStreamRedisSubscriber(subscriberJedis, responseWriter, writerList, subscriptionData); try { subscribeToChannel(executorService, aidrSubscriber, subscriptionData); jedisConn.setJedisSubscription(subscriberJedis, subscriptionData.patternSubscriptionFlag); } catch (Exception e) { logger.error(channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); OutputErrorHandler.sendErrorMail(e.getLocalizedMessage(), channelCode + ": Fatal exception occurred attempting subscription: " + e.toString()); } logger.info(channelCode + ": Spawning async response thread"); try { executorService.execute(aidrSubscriber); } catch (RejectedExecutionException|NullPointerException e) { logger.error(channelCode + "Fatal error executing async thread! Terminating."); } } } else { JsonObject errorMessage = new JsonObject(); errorMessage.addProperty("crisisCode", "NOT PROVIDED"); errorMessage.addProperty("streaming status", "ERROR"); if (callbackName != null) { responseWriter.write(callbackName + "(" + errorMessage.getAsString() + ")"); } else { responseWriter.write(errorMessage.getAsString()); } } return responseWriter; }  @Override void contextInitialized(ServletContextEvent sce); @Override void contextDestroyed(ServletContextEvent sce); String setFullyQualifiedChannelName(final String channelPrefixCode, final String channelCode); @GET @Path("/channel/{crisisCode}") @Produces(MediaType.APPLICATION_JSON) ChunkedOutput<String> streamChunkedResponse(
			@PathParam("crisisCode") String channelCode,
			@QueryParam("callback") final String callbackName,
			@DefaultValue("-1") @QueryParam("rate") Float rate,
			@DefaultValue("-1") @QueryParam("duration") String duration); static String redisHost; static int redisPort; }