Как создать исключение в функции из другого потока?

Question

Я столкнулся с очень странной проблемой: по сути, я реализую сервер потоковой передачи grp c в java и использую микропрофили в качестве библиотеки / структуры. Микропрофиль имеет очень удобные перехватчики, когда дело доходит до автоматических c повторов и аварийных откатов, но это зависит от исключений, генерируемых в аннотированной функции перехвата, в противном случае повторные попытки не запускаются.

Для унарного grp c звонки, это работает просто отлично, работает очень похоже на обычный вызов REST. Но когда клиент заставляет grp c поток на стороне сервера, protobuf создаст свой собственный поток и запросит обратный вызов для обработки onNext, onError и onCompleted. Поэтому, когда onError вызывается в области потока, любое выброшенное исключение не будет отправлено обратно какой-либо функции, использованной для запуска потока, поэтому нет исключений, вызывающих @ Retry.

Невозможно изменить способ обработки асинхронного потока, и невозможно изменить способ запуска @Retry. Grp c является сгенерированным кодом, а триггер @Retry основан на библиотеке микропрофилей.

Пример:

  @Retry(
          retryOn = {IOException.class, TimeoutException.class, StatusRuntimeException.class},
          maxDuration = 10,
          durationUnit =  ChronoUnit.SECONDS,
          maxRetries = 1,
          delay = 10,
          delayUnit = ChronoUnit.SECONDS
  )
  public void subscribeToLocations() {
    // --> Throwing an exception here triggers the @Retry <--
    SubscribeRequest locSubscribeRequest = SubscribeRequest.newBuilder().build();
    streamObserver = grpcStreamHandler();
    grpcBlockingstub.subscribeServerStreaming(locSubscribeRequest, streamObserver); // Can't change this.
  }

  private StreamObserver<SubscribeResponse> grpcStreamHandler() {
    return new StreamObserver<SubscribeResponse>() {
      @Override
      public void onNext(SubscribeResponse value) {
        // Handle grpc response
      }

      @Override
      public void onError(Throwable t) {
        // --> ERROR: Here, it should trigger the @Retry somehow. <--
      }

      @Override
      public void onCompleted() {
        // Handle oncomplete
      }
    };
  }

Я пытался найти решение этой проблемы дольше, чем Признаюсь, но я все еще в растерянности. Есть ли способ создать исключение в одной области, которая заканчивается в другой? Есть ли другое решение?

Answer 1

Здесь вы смешиваете операции syn c и asyn c, поэтому вам нужно четко определить, что для вас является «успехом», и вам не нужно повышать повтор. Это может зависеть от вашей логики c, нескольких вариантов с потоковой передачей:

• Получено как минимум одно обновление для onNext, затем завершение или ошибка
• onCompleted, полученная с обновлениями или без них
• et c

Как только вы определили правило «успеха», вы можете заключить подписку в Future и дождаться ее разрешения с предоставленным таймаутом. Если это не решено, выдается исключение из исходного вызова.