subprojects/parseq/src/main/java/com/linkedin/parseq/FusionTask.java
private <R> Consumer3<FusionTraceContext, Promise<S>, Settable<T>> compose(
      final Consumer3<FusionTraceContext, Promise<S>, Settable<R>> predecessor,
      final Consumer3<FusionTraceContext, Promise<R>, Settable<T>> propagator) {

    return (traceContext, src, dst) -> {

      /*
       * At this point we know that transformation chain's length > 1.
       * This code is executed during task execution, not when task is constructed.
       */
      traceContext.createSurrogate();

      predecessor.accept(traceContext, src, new Settable<R>() {

        @Override
        public void done(R value) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.value(value), dst);
          } catch (Exception e) {
            /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */
            LOGGER.error("ParSeq ingternal error. An exception was thrown by propagator", e);
          }
        }

        @Override
        public void fail(Throwable error) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.error(error), dst);
          } catch (Exception e) {
            /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */
            LOGGER.error("ParSeq ingternal error. An exception was thrown by propagator.", e);
          }
        }
      });
    };
  }
private <R> Consumer3<FusionTraceContext, Promise<S>, Settable<T>> compose(
      final Consumer3<FusionTraceContext, Promise<S>, Settable<R>> predecessor,
      final Consumer3<FusionTraceContext, Promise<R>, Settable<T>> propagator) {

    return (traceContext, src, dst) -> {

      /*
       * At this point we know that transformation chain's length > 1.
       * This code is executed during task execution, not when task is constructed.
       */
      traceContext.createSurrogate();

      predecessor.accept(traceContext, src, new Settable<R>() {

        @Override
        public void done(R value) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.value(value), dst);
          } catch (Exception e) {
            /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */
            LOGGER.error("ParSeq ingternal error. An exception was thrown by propagator", e);
          }
        }

        @Override
        public void fail(Throwable error) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.error(error), dst);
          } catch (Exception e) {
            /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */
            LOGGER.error("ParSeq ingternal error. An exception was thrown by propagator.", e);
          }
        }
      });
    };
  }
private < R > Consumer3 < FusionTraceContext , Promise < S > , Settable < T > > compose ( final Consumer3 < FusionTraceContext , Promise < S > , Settable < R > > predecessor , final Consumer3 < FusionTraceContext , Promise < R > , Settable < T > > propagator ) { return ( traceContext , src , dst ) -> { /*
       * At this point we know that transformation chain's length > 1.
       * This code is executed during task execution, not when task is constructed.
       */ traceContext . createSurrogate ( ) ; predecessor . accept ( traceContext , src , new Settable < R > ( ) { @ Override public void done ( R value ) throws PromiseResolvedException { try { /* Track start time of the transformation. End time is tracked by closure created by completing() */ getEffectiveShallowTraceBuilder ( traceContext ) . setStartNanos ( System . nanoTime ( ) ) ; propagator . accept ( traceContext , Promises . value ( value ) , dst ) ; } catch ( Exception e ) { /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */ LOGGER . error ( "ParSeq ingternal error. An exception was thrown by propagator" , e ) ; } } @ Override public void fail ( Throwable error ) throws PromiseResolvedException { try { /* Track start time of the transformation. End time is tracked by closure created by completing() */ getEffectiveShallowTraceBuilder ( traceContext ) . setStartNanos ( System . nanoTime ( ) ) ; propagator . accept ( traceContext , Promises . error ( error ) , dst ) ; } catch ( Exception e ) { /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */ LOGGER . error ( "ParSeq ingternal error. An exception was thrown by propagator." , e ) ; } } } ) ; } ; }
Composes transformation with the transformation of the predecessor.
Composes transformation with the transformation of the predecessor .