SiriClientSubscriptionManager implements StatusProviderService { synchronized void upgradePendingSubscription( SubscriptionResponseStructure response, StatusResponseStructure status, SubscriptionId subscriptionId, ESiriModuleType moduleType, SiriClientRequest originalSubscriptionRequest) { _log.info("upgrade pending subscription: {}", originalSubscriptionRequest); ClientSubscriptionChannel channel = getChannelForServer( originalSubscriptionRequest.getTargetUrl(), originalSubscriptionRequest.getTargetVersion()); ScheduledFuture<?> expiration = registerSubscriptionExpirationTask( subscriptionId, status, originalSubscriptionRequest); ClientSubscriptionInstance instance = new ClientSubscriptionInstance( channel, subscriptionId, originalSubscriptionRequest, moduleType, expiration); ClientSubscriptionInstance existing = _activeSubscriptions.put( subscriptionId, instance); if (existing != null) { _log.info("overwriting existing subscription: " + subscriptionId); } Set<SubscriptionId> channelSubscriptions = channel.getSubscriptions(); channelSubscriptions.add(subscriptionId); updateChannelWithClientRequest(channel, originalSubscriptionRequest, response); }  @Inject void setSchedulingService(SchedulingService schedulingService); @Inject void setHeartbeatManager(HeartbeatManager heartbeatManager); void registerPendingSubscription(SiriClientRequest request,
      SubscriptionRequest subscriptionRequest); void clearPendingSubscription(SiriClientRequest request,
      SubscriptionRequest subscriptionRequest); void handleSubscriptionResponse(SubscriptionResponseStructure response); boolean isSubscriptionActive(SubscriptionId subscriptionId); boolean isSubscriptionActiveForModuleDelivery(
      AbstractServiceDeliveryStructure moduleDelivery); void recordServiceDeliveryStatistics(ServiceDelivery serviceDelivery); SiriChannelInfo getChannelInfoForServiceDelivery(
      ServiceDelivery serviceDelivery); void handleTerminateSubscriptionResponse(
      TerminateSubscriptionResponseStructure response); void terminateAllSubscriptions(
      boolean waitForTerminateSubscriptionResponseOnExit); void handleCheckStatusNotification(
      CheckStatusResponseStructure response); void handleHeartbeatNotification(
      HeartbeatNotificationStructure heartbeat); @Override void getStatus(Map<String, String> status);  }
@Test public void testUpgradePendingSubscription() { SubscriptionResponseStructure response = new SubscriptionResponseStructure(); StatusResponseStructure status = new StatusResponseStructure(); SubscriptionId subscriptionId = new SubscriptionId("subscriberA", "subscriptionB"); ESiriModuleType moduleType = ESiriModuleType.VEHICLE_MONITORING; SiriClientRequest originalSubscriptionRequest = new SiriClientRequest(); originalSubscriptionRequest.setTargetUrl("http: _manager.upgradePendingSubscription(response, status, subscriptionId, moduleType, originalSubscriptionRequest); assertTrue(_manager.isSubscriptionActive(subscriptionId)); }
@Test public void testUpgradePendingSubscriptionWithExpirationTask() { SubscriptionResponseStructure response = new SubscriptionResponseStructure(); StatusResponseStructure status = new StatusResponseStructure(); SubscriptionId subscriptionId = new SubscriptionId("subscriberA", "subscriptionB"); ESiriModuleType moduleType = ESiriModuleType.VEHICLE_MONITORING; SiriClientRequest originalSubscriptionRequest = new SiriClientRequest(); originalSubscriptionRequest.setTargetUrl("http: originalSubscriptionRequest.setInitialTerminationDuration(60 * 1000); _manager.upgradePendingSubscription(response, status, subscriptionId, moduleType, originalSubscriptionRequest); Mockito.verify(_schedulingService).schedule(Mockito.any(Runnable.class), Mockito.eq(60 * 1000L), Mockito.eq(TimeUnit.MILLISECONDS)); }
@Test public void testUpgradePendingSubscriptionWithExpirationTaskAndResponseValidUntil() { SubscriptionResponseStructure response = new SubscriptionResponseStructure(); StatusResponseStructure status = new StatusResponseStructure(); status.setValidUntil(new Date(System.currentTimeMillis() + 30 * 1000L)); SubscriptionId subscriptionId = new SubscriptionId("subscriberA", "subscriptionB"); ESiriModuleType moduleType = ESiriModuleType.VEHICLE_MONITORING; SiriClientRequest originalSubscriptionRequest = new SiriClientRequest(); originalSubscriptionRequest.setTargetUrl("http: originalSubscriptionRequest.setInitialTerminationDuration(60 * 1000); _manager.upgradePendingSubscription(response, status, subscriptionId, moduleType, originalSubscriptionRequest); ArgumentCaptor<Long> timeoutCaptor = ArgumentCaptor.forClass(Long.class); Mockito.verify(_schedulingService).schedule(Mockito.any(Runnable.class), timeoutCaptor.capture(), Mockito.eq(TimeUnit.MILLISECONDS)); long timeout = timeoutCaptor.getValue(); long delta = Math.abs(30 * 1000L - timeout); assertTrue(delta < 150); }
@Test public void testExpirationTask() { SubscriptionResponseStructure response = new SubscriptionResponseStructure(); StatusResponseStructure status = new StatusResponseStructure(); SubscriptionId subscriptionId = new SubscriptionId("subscriberA", "subscriptionB"); ESiriModuleType moduleType = ESiriModuleType.VEHICLE_MONITORING; SiriClientRequest originalSubscriptionRequest = new SiriClientRequest(); originalSubscriptionRequest.setTargetUrl("http: originalSubscriptionRequest.setInitialTerminationDuration(60 * 1000); _manager.upgradePendingSubscription(response, status, subscriptionId, moduleType, originalSubscriptionRequest); ArgumentCaptor<Runnable> expirationTaskCaptor = ArgumentCaptor.forClass(Runnable.class); Mockito.verify(_schedulingService).schedule(expirationTaskCaptor.capture(), Mockito.eq(60 * 1000L), Mockito.eq(TimeUnit.MILLISECONDS)); Runnable expirationTask = expirationTaskCaptor.getValue(); expirationTask.run(); Mockito.verify(_terminateSubscriptionsManager).requestTerminationOfSubscription( Mockito.any(ClientSubscriptionInstance.class), Mockito.eq(true)); }
@Test public void testCheckStatusTask() { SubscriptionResponseStructure response = new SubscriptionResponseStructure(); StatusResponseStructure status = new StatusResponseStructure(); SubscriptionId subscriptionId = new SubscriptionId("subscriberA", "subscriptionB"); ESiriModuleType moduleType = ESiriModuleType.VEHICLE_MONITORING; SiriClientRequest originalSubscriptionRequest = new SiriClientRequest(); originalSubscriptionRequest.setTargetUrl("http: originalSubscriptionRequest.setCheckStatusInterval(30); _manager.upgradePendingSubscription(response, status, subscriptionId, moduleType, originalSubscriptionRequest); ArgumentCaptor<ClientSubscriptionChannel> captor = ArgumentCaptor.forClass(ClientSubscriptionChannel.class); Mockito.verify(_checkStatusManager).resetCheckStatusTask(captor.capture(), Mockito.eq(30)); ClientSubscriptionChannel channel = captor.getValue(); assertNull(channel.getCheckStatusUrl()); }
@Test public void testCheckStatusTaskWithCustomUrl() { SubscriptionResponseStructure response = new SubscriptionResponseStructure(); StatusResponseStructure status = new StatusResponseStructure(); SubscriptionId subscriptionId = new SubscriptionId("subscriberA", "subscriptionB"); ESiriModuleType moduleType = ESiriModuleType.VEHICLE_MONITORING; SiriClientRequest originalSubscriptionRequest = new SiriClientRequest(); originalSubscriptionRequest.setTargetUrl("http: originalSubscriptionRequest.setCheckStatusInterval(30); originalSubscriptionRequest.setCheckStatusUrl("http: _manager.upgradePendingSubscription(response, status, subscriptionId, moduleType, originalSubscriptionRequest); ArgumentCaptor<ClientSubscriptionChannel> captor = ArgumentCaptor.forClass(ClientSubscriptionChannel.class); Mockito.verify(_checkStatusManager).resetCheckStatusTask(captor.capture(), Mockito.eq(30)); ClientSubscriptionChannel channel = captor.getValue(); assertEquals("http: }