@@ -492,11 +492,11 @@ export class Worker {
492492 ...compiledOptions ,
493493 ...( compiledOptions . workflowBundle && isCodeBundleOption ( compiledOptions . workflowBundle )
494494 ? {
495- // Avoid dumping workflow bundle code to the console
496- workflowBundle : < WorkflowBundle > {
497- code : `<string of length ${ compiledOptions . workflowBundle . code . length } ,
498- } ,
499- }
495+ // Avoid dumping workflow bundle code to the console
496+ workflowBundle : < WorkflowBundle > {
497+ code : `<string of length ${ compiledOptions . workflowBundle . code . length } ,
498+ } ,
499+ }
500500 : { } ) ,
501501 } ,
502502 } ) ;
@@ -710,7 +710,7 @@ export class Worker {
710710 ) {
711711 logger . warn (
712712 'Ignoring WorkerOptions.interceptors.workflowModules because WorkerOptions.workflowBundle is set.\n' +
713- 'To use workflow interceptors with a workflowBundle, pass them in the call to bundleWorkflowCode.'
713+ 'To use workflow interceptors with a workflowBundle, pass them in the call to bundleWorkflowCode.'
714714 ) ;
715715 }
716716
@@ -883,8 +883,8 @@ export class Worker {
883883 */
884884 protected pollLoop$ < T > ( pollFn : ( ) => Promise < T > ) : Observable < T > {
885885 return from (
886- ( async function * ( ) {
887- for ( ; ; ) {
886+ ( async function * ( ) {
887+ for ( ; ; ) {
888888 try {
889889 yield await pollFn ( ) ;
890890 } catch ( err ) {
@@ -919,14 +919,14 @@ export class Worker {
919919 // so it can be cancelled if requested
920920 let output :
921921 | {
922- type : 'result' ;
923- result : coresdk . activity_result . IActivityExecutionResult ;
924- }
922+ type : 'result' ;
923+ result : coresdk . activity_result . IActivityExecutionResult ;
924+ }
925925 | {
926- type : 'run' ;
927- activity : Activity ;
928- input : ActivityExecuteInput ;
929- }
926+ type : 'run' ;
927+ activity : Activity ;
928+ input : ActivityExecuteInput ;
929+ }
930930 | { type : 'ignore' } ;
931931 switch ( variant ) {
932932 case 'start' : {
@@ -1112,38 +1112,44 @@ export class Worker {
11121112 */
11131113 protected nexusOperator ( ) : OperatorFunction < NexusTaskWithBase64Token , Uint8Array > {
11141114 return pipe (
1115- mergeMap ( async ( { task, base64TaskToken, protobufEncodedTask } ) : Promise < coresdk . nexus . INexusTaskCompletion | undefined > => {
1116- const { variant } = task ;
1117- if ( ! variant ) {
1118- throw new TypeError ( 'Got a nexus task without a "variant" attribute' ) ;
1119- }
1120-
1121- switch ( variant ) {
1122- case 'task' : {
1123- if ( task . task == null ) {
1124- throw new IllegalStateError ( `Got empty task for task variant with token: ${ base64TaskToken } ) ;
1125- }
1126- return await this . handleNexusRunTask ( task . task , base64TaskToken , protobufEncodedTask ) ;
1115+ mergeMap (
1116+ async ( {
1117+ task,
1118+ base64TaskToken,
1119+ protobufEncodedTask,
1120+ } ) : Promise < coresdk . nexus . INexusTaskCompletion | undefined > => {
1121+ const { variant } = task ;
1122+ if ( ! variant ) {
1123+ throw new TypeError ( 'Got a nexus task without a "variant" attribute' ) ;
11271124 }
1128- case 'cancelTask' : {
1129- const nexusHandler = this . taskTokenToNexusHandler . get ( base64TaskToken ) ;
1130- if ( nexusHandler == null ) {
1131- this . logger . trace ( 'Tried to cancel a non-existing nexus handler' , {
1132- taskToken : base64TaskToken ,
1133- } ) ;
1134- break ;
1125+
1126+ switch ( variant ) {
1127+ case 'task' : {
1128+ if ( task . task == null ) {
1129+ throw new IllegalStateError ( `Got empty task for task variant with token: ${ base64TaskToken } ` ) ;
1130+ }
1131+ return await this . handleNexusRunTask ( task . task , base64TaskToken , protobufEncodedTask ) ;
11351132 }
1136- // NOTE: nexus handler will not be considered cancelled until it confirms cancellation (by throwing a CancelledFailure)
1137- this . logger . trace ( 'Cancelling nexus handler' , nexusLogAttributes ( nexusHandler . info ) ) ;
1138- let reason = 'unkown' ;
1139- if ( task . cancelTask ?. reason != null ) {
1140- reason = coresdk . nexus . NexusTaskCancelReason [ task . cancelTask . reason ] ;
1133+ case 'cancelTask' : {
1134+ const nexusHandler = this . taskTokenToNexusHandler . get ( base64TaskToken ) ;
1135+ if ( nexusHandler == null ) {
1136+ this . logger . trace ( 'Tried to cancel a non-existing nexus handler' , {
1137+ taskToken : base64TaskToken ,
1138+ } ) ;
1139+ break ;
1140+ }
1141+ // NOTE: nexus handler will not be considered cancelled until it confirms cancellation (by throwing a CancelledFailure)
1142+ this . logger . trace ( 'Cancelling nexus handler' , nexusLogAttributes ( nexusHandler . info ) ) ;
1143+ let reason = 'unkown' ;
1144+ if ( task . cancelTask ?. reason != null ) {
1145+ reason = coresdk . nexus . NexusTaskCancelReason [ task . cancelTask . reason ] ;
1146+ }
1147+ nexusHandler . abortController . abort ( new CancelledFailure ( reason ) ) ;
1148+ return { ackCancel : true , taskToken : task . cancelTask ?. taskToken } ;
11411149 }
1142- nexusHandler . abortController . abort ( new CancelledFailure ( reason ) ) ;
1143- return { ackCancel : true , taskToken : task . cancelTask ?. taskToken }
11441150 }
11451151 }
1146- } ) ,
1152+ ) ,
11471153 filter ( < T > ( result : T ) : result is Exclude < T , undefined > => result !== undefined ) ,
11481154 map ( ( result ) => coresdk . nexus . NexusTaskCompletion . encodeDelimited ( result ) . finish ( ) )
11491155 ) ;
@@ -1278,9 +1284,9 @@ export class Worker {
12781284 const completion = synthetic
12791285 ? undefined
12801286 : coresdk . workflow_completion . WorkflowActivationCompletion . encodeDelimited ( {
1281- runId : activation . runId ,
1282- successful : { } ,
1283- } ) . finish ( ) ;
1287+ runId : activation . runId ,
1288+ successful : { } ,
1289+ } ) . finish ( ) ;
12841290 return { state : undefined , output : { close, completion } } ;
12851291 }
12861292
0 commit comments