SubscriptionService extends SubscriptionServiceGrpc.SubscriptionServiceImplBase implements Logging { @Override public void activate(Subscription subscription, StreamObserver<SubscriptionUpdate> observer) { _debug().log("Activating the subscription: `%s`.", subscription); try { SubscriptionCallback callback = forwardingTo(observer); StreamObserver<Response> responseObserver = forwardErrorsOnly(observer); Optional<BoundedContext> foundContext = findContextOf(subscription); if (foundContext.isPresent()) { Stand targetStand = foundContext.get().stand(); targetStand.activate(subscription, callback, responseObserver); } else { for (BoundedContext context : typeToContextMap.values()) { Stand stand = context.stand(); stand.activate(subscription, callback, responseObserver); } } } catch (@SuppressWarnings("OverlyBroadCatchBlock") Exception e) { _error().withCause(e) .log("Error activating the subscription."); observer.onError(e); } } private  SubscriptionService(ImmutableMap<TypeUrl, BoundedContext> map); static Builder newBuilder(); @Override void subscribe(Topic topic, StreamObserver<Subscription> responseObserver); @Override void activate(Subscription subscription, StreamObserver<SubscriptionUpdate> observer); @Override void cancel(Subscription subscription, StreamObserver<Response> responseObserver);  }
@Test @MuteLogging @DisplayName("receive IAE in error callback when activating non-existent subscription") void failOnActivatingNonExistent() { Subscription invalidSubscription = invalidSubscription(); subscriptionService.activate(invalidSubscription, activationObserver); assertThat(activationObserver.responses()) .isEmpty(); assertThat(activationObserver.isCompleted()) .isFalse(); assertThat(activationObserver.getError()) .isInstanceOf(InvalidSubscriptionException.class); }