class documentation

class ActivityInboundInterceptor: (source)

View In Hierarchy

Inbound interceptor to wrap outbound creation and activity execution.

This should be extended by any activity inbound interceptors.

Method __init__ Create the inbound interceptor.
Async Method execute_activity Called to invoke the activity.
Method init Initialize with an outbound interceptor.
Instance Variable next Undocumented
def __init__(self, next: ActivityInboundInterceptor): (source)

Create the inbound interceptor.

Parameters
next:ActivityInboundInterceptorThe next interceptor in the chain. The default implementation of all calls is to delegate to the next interceptor.
async def execute_activity(self, input: ExecuteActivityInput) -> Any: (source)

Called to invoke the activity.

def init(self, outbound: ActivityOutboundInterceptor): (source)

Initialize with an outbound interceptor.

To add a custom outbound interceptor, wrap the given interceptor before sending to the next init call.

next = (source)

Undocumented