GrpcStreamSubscription implements hera.api.model.Subscription<T> { @Override public void unsubscribe() { if (!context.isCancelled()) { synchronized (lock) { if (!context.isCancelled()) { logger.debug("Unsubscribe subscription"); context.close(); } else { logger.debug("Subscription is already cancelled"); } } } else { logger.debug("Subscription is already cancelled"); } }  @Override void unsubscribe(); @Override boolean isUnsubscribed(); @Override String toString();  }
@Test public void testUnsubscribe() { final io.grpc.Context.CancellableContext context = io.grpc.Context.current().withCancellation(); context.run(new Runnable() { @Override public void run() { } }); final GrpcStreamSubscription<Object> subscription = new GrpcStreamSubscription<>(context); subscription.unsubscribe(); assertTrue(subscription.isUnsubscribed()); }