私のスタブは多くのメソッドを使用しています。チャネルを再作成して使用するたびにチャネルを閉じることはできません。チャネルはランダムであるため、すべてを配置することはできません。そのため、タイムアウト時に作成したチャネルが自動的に閉じられるようにしたいと考えています。自分のルート メソッド用のチャネル、およびルート メソッドを使用する他のチャネルを作成します。このルート メソッドが使用されなくなった場合、またはこのルート メソッドが使用されていない場合、チャネルを自動的にオフにできるようにしたいと考えています。長時間使用すると、自動的にチャンネルがオフになります
@staticmethod
def _create_client(workspace):
workspace._client = TreeLabClient(token=workspace.token)
def __init__(self, token: str):
self.metadata = [('access_token', token)]
if not os.getenv('ip'):
os.environ['ip'] = get_env_ip()
self.channel = grpc.insecure_channel(os.getenv('ip'))
self.client = treelab_service.TreeLabApiServiceStub(self.channel)
self.connectivity_state = _ChannelConnectivityState(self.channel)
self.stub = treelab_service.TreeLabApiServiceStub(self.channel)
def close(self):
self.channel.close()
@wait("WorkspaceCreated")
def create_workspace(self, create_workspace_input: CreateWorkspaceInput):
return self.stub.CreateWorkspace, create_workspace_input, self.metadata
@wait("WorkspaceUpdated")
def update_workspace(self, update_workspace_input: UpdateWorkspaceInput):
return self.stub.UpdateWorkspace, update_workspace_input, self.metadata
@wait("CoreCreated")
def add_core(self, add_core_input):
return self.stub.AddCore, add_core_input, self.metadata
@wait("CoreRemoved")
def remove_core(self, remove_core_input: RemoveCoreInput):
return self.stub.RemoveCore, remove_core_input, self.metadata
@wait("CoreUpdated")
def update_core(self, update_core_input: UpdateCoreInput):
return self.stub.UpdateCore, update_core_input, self.metadata
@wait("TableCreated")
def add_table(self, add_table_input: AddTableInput):
return self.stub.AddTable, add_table_input, self.metadata
GRPC のチャネルに有効期限を設定できるようにしたい