Почему функциональность Observable выполняется дважды за один вызов?

Question

Полная структура программы

Аннотация:

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface UserAnnotation {
}

Затем создан Перехватчик:

public class UserInterceptor implements MethodInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(UserInterceptor.class);

    @Inject
    UserService userService; // this is not working

    public Object invoke(MethodInvocation invocation) throws Throwable {
        logger.info("UserInterceptor : Interceptor Invoked");
        Object result = invocation.proceed();
        Observable<List<User>> observable = (Observable<List<Sample>>) result;
        observable.flatMap(Observable::from).subscribe(object -> {
            User user = (User)object
            SampleSender sender = new SampleSender();
            sender.setBoolean(user.isBoolean());
            logger.info("Pushing Data into Sender");
            userService.insert(String.join("_", "key", "value"), sender); 
        }
        return result;
    }
}

Затем я создал GuiceModule, как показано ниже: -

public class UserModule extends AbstractModule {
    @Override
    protected void configure() {
        SampleInterceptor interceptor = new SampleInterceptor()
        requestInjection(interceptor);
        bindInterceptor(Matchers.any(), Matchers.annotatedWith(SampleAnnotation.class), interceptor);
}

}

Класс, в котором я использую вышеупомянутую аннотацию:

// This class also have so many method and this was already declared and using in another services, I created a sample class here
class UserClassForInterceptor {

      @Inject
      AnotherClass anotherClass;
      // this userMethod() is not a new method, its already created, 
      // now I am adding annotation to it, because after finishing this functionality, 
      // I want something should be done, so created annotation and added here
      @UserAnnotation
      public Observable<List<Sample>> userMethod() {
            logger.info("This is printing only once");
            return anotherClass.getUser().flatMap(user ->{
                logger.info("This is also printing twice");
                // this logger printed twise means, this code snippet is getting executed twise
            });
      }
}

public class AnotherClass{
          public Observable<User> getUser(){
           Observable<Sample> observableSample = methodReturnsObservableSample();
           logger.info("Getting this logger only once");
           return observableSample.map(response-> {
               logger.info("This logger is printing twice");
               //here have code to return observable of User
           });
      }
}

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

У меня есть RestModule, с помощью которого я связываю UserClassForInterceptor следующим образом

public final class RestModule extends JerseyServletModule {
    // other classes binding
    bind(UserClassForInterceptor.class).in(Scopes.SINGLETON);
    // other classes binding
    install(new SampleModule());
}

Теперь у меня есть класс начальной загрузки, в котором яобязателен RestModule

public class Bootstrap extends ServerBootstrap {
   binder.install(new RestModule());
}

Использование: -

@Path("service/sample")
public class SampleRS {
    @Inject
    UserClassForInterceptor userClassForInterceptor;

    public void someMethod() {
        userClassForInterceptor.sampleMethod();
    }
}

Answer 1

Этот код также помог мне

public Object invoke(MethodInvocation invocation) throws Throwable {
    Object result = null;
    try{
        logger.debug("Interceptor Invoked");
        result = invocation.proceed();
        Observable<List<User>> observable = (Observable<List<User>>)result;

        return observable
                .doOnNext(this::updateUser);
    }
    catch(Exception ex){
        logger.error("Error: ",ex);
    }
    return result;
}

private void updateUser(List<User> users) {
    if(CollectionUtils.isNotEmpty(users)) {
        for(User user: users) {
            SampleSender sender = new SampleSender();
            sender.setBoolean(user.isBoolean());
            logger.info("Pushing Data into Sender");
            userService.insert(String.join("_", "key", "value"), sender); 
        }
    }
}

Answer 2

Вы создали аннотацию @UserAnnotation и класс-перехватчик для добавления аннотации.Вы присоединяете аннотацию к методу userMethod().

Первое, что делает ваша подпрограмма-перехватчик, - это вызывает userMethod(), чтобы получить наблюдаемое, которое она возвращает, и затем перехватчик подписывается навозвращается наблюдаемым, в результате чего появляются первые сообщения журнала.В конце концов, перехватчик возвращает наблюдаемое исходному вызывающему.Когда что-то еще присоединяется к возвращаемой наблюдаемой, цепочка наблюдателей активируется второй раз, поэтому сообщения журнала появляются дважды.

RxJava имеет побочные эффекты

Пока RxJava являетсяреализация концепции «функционально-реактивного программирования», цепочки наблюдателей, которые вы строите (функционально), работают только тогда, когда на них подписаны, и эти подписки имеют побочные эффекты.Регистрация выходных данных является одним из побочных эффектов и, вероятно, наиболее благоприятным;изменения в переменных или вызовы методов, которые имеют побочные эффекты, оказывают более широкое влияние.

Когда построена цепочка наблюдателей (правильно), она действует как потенциальное вычисление, пока не будет подписчик.Если вам нужно иметь более одного подписчика, как, например, для проблемного домена, вам нужно решить, нужно ли активировать цепочку наблюдателей для каждой подписки, обычного случая или только один раз для всех перекрывающихся подписок.

Если вы хотите, чтобы у всех перекрывающихся подписок была одна и та же наблюдаемая, вы можете использовать оператор share().Существует ряд связанных операторов, которые влияют на время существования наблюдаемых и подписок.Вот краткий обзор: Как использовать оператор RxJava share ()?

Аспектно-ориентированное программирование: перехватчики и Guice

Ваш код использует Guiceпредоставить возможность под названием «аспектно-ориентированное программирование».Это позволяет вам вводить код в вашу программу для решения сквозных задач или для повышения ее функциональности путем настройки управляемых шлюзов.Использование Guice или аналогичных подходов AOP требует дисциплины .

. В вашем случае вы использовали процесс перехвата, чтобы вызвать необъяснимые (до сих пор) побочные эффекты, подписавшись на цепочку наблюдателей, у которой неттривиальные побочные эффекты.Представьте, что метод, который вы перехватили, установил одноразовое соединение, и ваш перехватчик использовал это соединение, выполняя свою работу, в результате чего исходный вызывающий абонент не смог использовать соединение.

Дисциплина, в которой вы нуждаетесь, заключается в понимании правилчто перехватчик должен следовать.Подумайте о таких правилах, как «Во-первых, не навреди».

Doing Things The FRP Way

Если вам нужно добавить дополнительный шаг при обработке информации пользователя, тогдавы должны создать новую наблюдаемую в вашем перехватчике, которая делает это, но только тогда, когда исходный вызывающий абонент подписался на наблюдаемую:

    Object result = invocation.proceed();
    Observable<List<User>> observable = (Observable<List<Sample>>) result;
    Observable<List<User>> newObservable = observable
      .doOnNext( sampleList ->
         Observable.fromIterable( sampleList )
           .subscribe(object -> {
             User user = (User)object
             SampleSender sender = new SampleSender();
             sender.setBoolean(user.isBoolean());
             logger.info("Pushing Data into Sender");
             userService.insert(String.join("_", "key", "value"), sender); 
           }));
    return newObservable;

Возвращая модифицированную цепочку наблюдателей, вы не вводите побочные эффекты от оригинала.цепочка наблюдателей и убедитесь, что побочные эффекты, которые вы вводите в свой собственный код, будут только запускаться при подписке на исходную цепочку наблюдателей.