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 .