Plugins to support controlling multiple configuration points at once

[tconley1428]

Is your feature request related to a problem? Please describe.

Users often need to configure a number of things in order to support certain systems. For instance, all workflows at Company X might need to register a dataconverter, customize the sandbox, register a set of activities, etc.

Describe the solution you'd like

A plugin model to allow configuring all these points at once.

# In client (would rather have this on client.Interceptor, but people
# are concerned with the name "interceptor" though I prefer it)
class Plugin:
  # NOTE: ClientConnectConfig could help at https://github.com/temporalio/sdk-python/pull/895
  async def connect_client(self, config: ClientConnectConfig, next: Plugin) -> Client:
    # Default impl calls next
	  
  def create_client(self, config: ClientConfig, next: Plugin) -> Client:
    # Default impl calls next
    
# In worker (would rather have this on worker.Interceptor, but people
# are concerned with the name "interceptor" though I prefer it)
class Plugin:
  def create_worker(self, config: WorkerConfig, next: Plugin) -> Worker:
    # Default impl calls next
    
  async def run_worker(self, worker: Worker, next: Plugin) -> None:
	  # Default impl calls next

For example:

class MyPlugin(client.Plugin, worker.Plugin, client.Interceptor, worker.Interceptor):
  def create_client(self, config: ClientCreateConfig, next: Plugin) -> Client:
    # Say you want to set data converter and interceptor
    config["data_converter"] = MyDataConverter()
    config["interceptors"].append(self)
    return super().create_client(config, next)
    
  def create_worker(self, config: WorkerCreateConfig, next: Plugin) -> Worker:
    # Say you want an activity
    config["activities"].append(MyActivity())
    return super().create_worker(config, next)
    
  async def run_worker(self, worker: Worker, next: Plugin) -> None:
    try:
      await super().run_worker(worker, next)
    finally:
      # Say you want to do something on worker shutdown
      do_some_cleanup()
      
  def intercept_activity(self, next: ActivityInboundInterceptor) -> ActivityInboundInterceptor:
    # Say you want to intercept activities
    return MyActivityInterceptor(next)

Additional context

Per-SDK Tickets

• Go - Plugin support sdk-go#2020
• Java - Plugin support sdk-java#2626
• Core - N/A
• TypeScript - [Feature Request] Plugin support sdk-typescript#1764
• Python - [Feature Request] Plugins to support controlling multiple configuration points at once sdk-python#950 and 💥 Replayer configuration from plugins sdk-python#1011
• .NET - [Feature Request] Plugin support sdk-dotnet#507
• Ruby - [Feature Request] Plugin support sdk-ruby#318
• PHP - [Feature Request] Plugin support sdk-php#644
• Temporal CLI - N/A

[cretz]

Note to implementers - there are also replayer behaviors that have to be accounted for in the worker plugin. The client options should accept client plugins (that also become worker plugins if they implement that interface too), and the worker options should accept worker plugins. Each language may have its challenges here, but the idea is to basically follow the interceptor model.