diff --git a/pyproject.toml b/pyproject.toml index a9492c59e..ebbdc02bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -188,6 +188,7 @@ exclude = [ 'temporalio/nexus/system/workflow_service', 'temporalio/contrib/streams/proto', 'temporalio/contrib/streams/system', + 'temporalio/contrib/streams/writer_system', ] [[tool.mypy.overrides]] @@ -195,13 +196,17 @@ module = "temporalio.nexus.system.workflow_service.*" ignore_errors = true [[tool.mypy.overrides]] -module = ["temporalio.contrib.streams.proto.*", "temporalio.contrib.streams.system.*"] +module = [ + "temporalio.contrib.streams.proto.*", + "temporalio.contrib.streams.system.*", + "temporalio.contrib.streams.writer_system.*", +] ignore_errors = true [tool.pydocstyle] convention = "google" # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088 -match_dir = "^(?!(docs|scripts|tests|api|proto|system|_support|notifications|\\.)).*" +match_dir = "^(?!(docs|scripts|tests|api|proto|system|writer_system|_support|notifications|\\.)).*" add_ignore = [ # We like to wrap at a certain number of chars, even long summary sentences. # https://github.com/PyCQA/pydocstyle/issues/184 diff --git a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto index 90d75ef86..575ece730 100644 --- a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto +++ b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto @@ -5,6 +5,13 @@ syntax = "proto3"; package temporal.api.streamservice.v1; +option go_package = "go.temporal.io/api/streamservice/v1;streamservice"; +option java_package = "io.temporal.api.streamservice.v1"; +option java_multiple_files = true; +option java_outer_classname = "StreamServiceProto"; +option ruby_package = "Temporalio::Api::StreamService::V1"; +option csharp_namespace = "Temporalio.Api.StreamService.V1"; + import "google/protobuf/duration.proto"; import "google/protobuf/timestamp.proto"; import "temporal/api/common/v1/message.proto"; diff --git a/scripts/_wit/stream-writer.wit b/scripts/_wit/stream-writer.wit new file mode 100644 index 000000000..8f021dab7 --- /dev/null +++ b/scripts/_wit/stream-writer.wit @@ -0,0 +1,98 @@ +// Internal. Not a Temporal API, and no server serves it. The crate's Worker extension serves +// these calls in process, backed by the messages in ../protos/temporal/sdk/streams/v1/ +// extension.proto. Lang reaches them through an extension-call transport, from Activities and +// clients only. So the interface is client-only and has no endpoint, which keeps it out of the +// System Nexus path. The writer keeps the Producer epoch and the sequence in the crate. This file +// is the source. Each SDK copies it beside stream-service.wit and generates its bindings with +// nexgen. +package temporal:streams@1.0.0; + +world writer { + export stream-writer; +} + +/// @nexus.service-name "temporal.sdk.streams.v1.StreamWriter" +/// @nexus.client-only +/// @nexus.experimental +interface stream-writer { + use nexus:temporal-types/model@1.0.0.{duration, placeholder}; + use temporal:nexus/stream-service@1.0.0.{append-record, producer, stream-ref}; + + /// @nexus.doc "Open a writer for one Producer of one Stream." + /// @nexus.proto "temporal.sdk.streams.v1.OpenWriterRequest" + record open-writer-request { + %stream: option, + producer: option, + /// @nexus.doc "A write sends at most this many Records in one batch. Zero is 1000." + max-batch-size: option, + /// @nexus.doc "How long a write retries a Store failure. Empty is 10 s." + max-retry-duration: option, + /// @nexus.omit + namespace: placeholder, + } + + /// @nexus.proto "temporal.sdk.streams.v1.WriteRecordsRequest" + record write-records-request { + writer-id: option, + records: list, + } + + /// @nexus.proto "temporal.sdk.streams.v1.WriteRecordsResponse" + record write-records-response { + /// @nexus.doc "The offset of the first Record. Zero when the request has no Record." + first-offset: option, + /// @nexus.doc "The offset of the last Record. Zero when the request has no Record." + last-offset: option, + } + + /// @nexus.proto "temporal.sdk.streams.v1.CloseWriterRequest" + record close-writer-request { + writer-id: option, + } + + /// @nexus.proto "temporal.sdk.streams.v1.CloseWriterResponse" + record close-writer-response { + /// @nexus.doc "The offset of the newest Record the writer wrote. Zero if it wrote none." + last-offset: option, + } + + /// @nexus.proto "temporal.sdk.streams.v1.CloseWritersRequest" + record close-writers-request { + } + + /// @nexus.proto "temporal.sdk.streams.v1.CloseWritersResponse" + record close-writers-response { + /// @nexus.doc "The writers that had failed." + failed: option, + } + + // A writer never changes. Offsets are values that its methods return. + /// @nexus.doc "An open writer of one Producer. The crate keeps its epoch and sequence." + resource writer { + constructor(writer-id: u64); + + /// @nexus.doc "Write Records at once. The answer has the first and the last offset." + /// @nexus.operation "write-records" + write-records: func(records: list) -> write-records-response; + + /// @nexus.doc "Wait for a write in flight, then drop the writer. The Stream stays open." + /// @nexus.operation "close-writer" + close: func() -> close-writer-response; + } + + /// @nexus.proto "temporal.sdk.streams.v1.OpenWriterResponse" + type writer-result = own; + + /// @nexus.operation name="OpenWriter" + open-writer: func(request: open-writer-request) -> writer-result; + + /// @nexus.operation name="WriteRecords" + write-records: func(request: write-records-request) -> write-records-response; + + /// @nexus.operation name="CloseWriter" + close-writer: func(request: close-writer-request) -> close-writer-response; + + /// @nexus.doc "Drop every open writer, for example at shutdown." + /// @nexus.operation name="CloseWriters" + close-writers: func(request: close-writers-request) -> close-writers-response; +} diff --git a/scripts/gen_stream_service.py b/scripts/gen_stream_service.py index 292303a52..473825f75 100644 --- a/scripts/gen_stream_service.py +++ b/scripts/gen_stream_service.py @@ -18,6 +18,7 @@ base_dir = Path(__file__).parent.parent wit_path = base_dir / "scripts" / "_wit" / "stream-service.wit" +writer_wit_path = base_dir / "scripts" / "_wit" / "stream-writer.wit" model_wit_path = ( gen_protos.api_proto_dir / "nexus" / "deps" / "nexus-temporal-types" / "model.wit" ) @@ -30,10 +31,12 @@ / "stream_service.proto" ) output_dir = base_dir / "temporalio" / "contrib" / "streams" / "system" +# The writer is client-only, so it has a caller and its resource class, and no Workflow code. +writer_output_dir = base_dir / "temporalio" / "contrib" / "streams" / "writer_system" NEX_GEN_REPOSITORY = "https://github.com/moetemp/nexgen" # The head of moe/brook/v3/n8-stream-resource. A branch moves, so the pin is the commit. -NEX_GEN_REVISION = "10fc35b40fcf48651c25e654a260bc71585f1429" +NEX_GEN_REVISION = "204b566ee996939b165ccd1d18c8c432bfdcc0dd" nex_gen_root = base_dir / ".nexgen-streams" / NEX_GEN_REVISION @@ -73,20 +76,44 @@ def api_protos() -> list[Path]: def generate_stream_service() -> None: - for path in (wit_path, model_wit_path, proto_path): + for path in (wit_path, writer_wit_path, model_wit_path, proto_path): if not path.exists(): raise RuntimeError(f"missing generator input: {path}") with tempfile.TemporaryDirectory(dir=base_dir) as temp_dir: descriptor_path = Path(temp_dir) / "stream_service.bin" + # The extension messages live in temporalio.contrib.streams.proto, so the descriptor + # names their file by that module path. + extension_root = Path(temp_dir) / "extension" + extension_path = ( + extension_root + / "temporalio" + / "contrib" + / "streams" + / "proto" + / "v1" + / "extension.proto" + ) + extension_path.parent.mkdir(parents=True) + shutil.copy( + gen_protos.additional_proto_dir + / "temporal" + / "sdk" + / "streams" + / "v1" + / "extension.proto", + extension_path, + ) subprocess.check_call( [ sys.executable, "-mgrpc_tools.protoc", f"--proto_path={gen_protos.api_proto_dir}", f"--proto_path={gen_protos.additional_proto_dir}", + f"--proto_path={extension_root}", "--include_imports", f"--descriptor_set_out={descriptor_path}", str(proto_path), + str(extension_path), # The model WIT names API types, so their descriptors must be in the set. *map(str, api_protos()), ] @@ -109,19 +136,38 @@ def generate_stream_service() -> None: str(output_dir), ] ) - subprocess.check_call( - [ - sys.executable, - "-m", - "ruff", - "check", - "--select", - "I", - "--fix", - str(output_dir), - ] - ) - subprocess.check_call([sys.executable, "-m", "ruff", "format", str(output_dir)]) + shutil.rmtree(writer_output_dir, ignore_errors=True) + subprocess.check_call( + [ + *nex_gen_command(), + "python", + str(writer_wit_path), + str(wit_path), + str(model_wit_path), + "--native-api", + "--client", + "--support-package", + "temporalio.nexus.system._support", + "--descriptors", + str(descriptor_path), + "--output", + str(writer_output_dir), + ] + ) + for directory in (output_dir, writer_output_dir): + subprocess.check_call( + [ + sys.executable, + "-m", + "ruff", + "check", + "--select", + "I", + "--fix", + str(directory), + ] + ) + subprocess.check_call([sys.executable, "-m", "ruff", "format", str(directory)]) if __name__ == "__main__": diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.py b/temporalio/api/streamservice/v1/stream_service_pb2.py index 5da897aa4..a2664ae30 100644 --- a/temporalio/api/streamservice/v1/stream_service_pb2.py +++ b/temporalio/api/streamservice/v1/stream_service_pb2.py @@ -26,7 +26,7 @@ ) DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n2temporal/api/streamservice/v1/stream_service.proto\x12\x1dtemporal.api.streamservice.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/api/failure/v1/message.proto"]\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x1e\n\x16\x66irst_execution_run_id\x18\x03 \x01(\t\x12\x0f\n\x07\x62\x61\x63kend\x18\x04 \x01(\t"\xc8\x01\n\rProducerEpoch\x12\x1d\n\x15task_started_event_id\x18\x01 \x01(\x03\x12\x14\n\x0ctask_attempt\x18\x02 \x01(\x05\x12\x35\n\x11task_started_time\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16\x61ttempt_scheduled_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x9e\x01\n\x08Producer\x12\x39\n\x04kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\n\n\x02id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12\x0e\n\x06run_id\x18\x04 \x01(\t"p\n\x11WorkflowTaskEpoch\x12\x18\n\x10started_event_id\x18\x01 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05\x12\x30\n\x0cstarted_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp"L\n\x0c\x41ppendRecord\x12\r\n\x05topic\x18\x01 \x01(\t\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"\xb3\x02\n\x0cSupersession\x12\x42\n\rproducer_kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12Q\n\x17previous_completed_task\x18\x04 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x05 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x06 \x01(\t\x12\x0e\n\x06run_id\x18\x07 \x01(\t"\x89\x03\n\x0cStreamRecord\x12=\n\x04kind\x18\x01 \x01(\x0e\x32/.temporal.api.streamservice.v1.StreamRecordKind\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x39\n\x08producer\x18\x04 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12-\n\x04\x62ody\x18\x05 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x41\n\x0csupersession\x18\x06 \x01(\x0b\x32+.temporal.api.streamservice.v1.Supersession\x12\x35\n\x0c\x63lose_result\x18\x07 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x37\n\rclose_failure\x18\x08 \x01(\x0b\x32 .temporal.api.failure.v1.Failure"\xae\x01\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x11\n\tstream_id\x18\x02 \x01(\t\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t\x12-\n\nauto_close\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration"\x9e\x01\n\x14\x43reateStreamResponse\x12\x38\n\x06stream\x18\x01 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\x12;\n\x06status\x18\x03 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus"\xed\x02\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12<\n\x07records\x18\x05 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x08 \x01(\t"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\xe9\x02\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12/\n\x06result\x18\x04 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x31\n\x07\x66\x61ilure\x18\x05 \x01(\x0b\x32 .temporal.api.failure.v1.Failure\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03"\x15\n\x13\x43loseStreamResponse"\xd9\x02\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x37\n\x05start\x18\x04 \x01(\x0e\x32(.temporal.api.streamservice.v1.ReadStart\x12\x0e\n\x06topics\x18\x05 \x03(\t\x12\x13\n\x0bmax_records\x18\x06 \x01(\x05\x12\'\n\x04wait\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12.\n\x0b\x63reate_wait\x18\x08 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0etopic_prefixes\x18\n \x03(\tJ\x04\x08\t\x10\nR\x0cidle_timeout"v\n\x13ReadRecordsResponse\x12<\n\x07records\x18\x01 \x03(\x0b\x32+.temporal.api.streamservice.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08"d\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef"\xc7\x01\n\x16\x44\x65scribeStreamResponse\x12;\n\x06status\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nauto_close\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration*\xc8\x01\n\x0cProducerKind\x12\x1d\n\x19PRODUCER_KIND_UNSPECIFIED\x10\x00\x12\x1a\n\x16PRODUCER_KIND_WORKFLOW\x10\x01\x12 \n\x1cPRODUCER_KIND_LOCAL_ACTIVITY\x10\x02\x12\x1a\n\x16PRODUCER_KIND_ACTIVITY\x10\x03\x12\x18\n\x14PRODUCER_KIND_CLIENT\x10\x04\x12%\n!PRODUCER_KIND_STANDALONE_ACTIVITY\x10\x05*\xb3\x01\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12!\n\x1dSTREAM_RECORD_KIND_SUPERSEDED\x10\x02\x12\x1c\n\x18STREAM_RECORD_KIND_CLOSE\x10\x03"\x04\x08\x04\x10\x04*\x17STREAM_RECORD_KIND_IDLE*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03*U\n\tReadStart\x12\x1a\n\x16READ_START_UNSPECIFIED\x10\x00\x12\x18\n\x14READ_START_BEGINNING\x10\x01\x12\x12\n\x0eREAD_START_END\x10\x02\x62\x06proto3' + b'\n2temporal/api/streamservice/v1/stream_service.proto\x12\x1dtemporal.api.streamservice.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a%temporal/api/failure/v1/message.proto"]\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\x12\x1e\n\x16\x66irst_execution_run_id\x18\x03 \x01(\t\x12\x0f\n\x07\x62\x61\x63kend\x18\x04 \x01(\t"\xc8\x01\n\rProducerEpoch\x12\x1d\n\x15task_started_event_id\x18\x01 \x01(\x03\x12\x14\n\x0ctask_attempt\x18\x02 \x01(\x05\x12\x35\n\x11task_started_time\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16\x61ttempt_scheduled_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x9e\x01\n\x08Producer\x12\x39\n\x04kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\n\n\x02id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12\x0e\n\x06run_id\x18\x04 \x01(\t"p\n\x11WorkflowTaskEpoch\x12\x18\n\x10started_event_id\x18\x01 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05\x12\x30\n\x0cstarted_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp"L\n\x0c\x41ppendRecord\x12\r\n\x05topic\x18\x01 \x01(\t\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"\xb3\x02\n\x0cSupersession\x12\x42\n\rproducer_kind\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.ProducerKind\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12;\n\x05\x65poch\x18\x03 \x01(\x0b\x32,.temporal.api.streamservice.v1.ProducerEpoch\x12Q\n\x17previous_completed_task\x18\x04 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x05 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x06 \x01(\t\x12\x0e\n\x06run_id\x18\x07 \x01(\t"\x89\x03\n\x0cStreamRecord\x12=\n\x04kind\x18\x01 \x01(\x0e\x32/.temporal.api.streamservice.v1.StreamRecordKind\x12\x0e\n\x06offset\x18\x02 \x01(\x03\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x39\n\x08producer\x18\x04 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12-\n\x04\x62ody\x18\x05 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x41\n\x0csupersession\x18\x06 \x01(\x0b\x32+.temporal.api.streamservice.v1.Supersession\x12\x35\n\x0c\x63lose_result\x18\x07 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x37\n\rclose_failure\x18\x08 \x01(\x0b\x32 .temporal.api.failure.v1.Failure"\xae\x01\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x11\n\tstream_id\x18\x02 \x01(\t\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t\x12-\n\nauto_close\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration"\x9e\x01\n\x14\x43reateStreamResponse\x12\x38\n\x06stream\x18\x01 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\x12;\n\x06status\x18\x03 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus"\xed\x02\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12<\n\x07records\x18\x05 \x03(\x0b\x32+.temporal.api.streamservice.v1.AppendRecord\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x08 \x01(\t"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\xe9\x02\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x39\n\x08producer\x18\x03 \x01(\x0b\x32\'.temporal.api.streamservice.v1.Producer\x12/\n\x06result\x18\x04 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x31\n\x07\x66\x61ilure\x18\x05 \x01(\x0b\x32 .temporal.api.failure.v1.Failure\x12Q\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32\x30.temporal.api.streamservice.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03"\x15\n\x13\x43loseStreamResponse"\xd9\x02\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x37\n\x05start\x18\x04 \x01(\x0e\x32(.temporal.api.streamservice.v1.ReadStart\x12\x0e\n\x06topics\x18\x05 \x03(\t\x12\x13\n\x0bmax_records\x18\x06 \x01(\x05\x12\'\n\x04wait\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12.\n\x0b\x63reate_wait\x18\x08 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0etopic_prefixes\x18\n \x03(\tJ\x04\x08\t\x10\nR\x0cidle_timeout"v\n\x13ReadRecordsResponse\x12<\n\x07records\x18\x01 \x03(\x0b\x32+.temporal.api.streamservice.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08"d\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x38\n\x06stream\x18\x02 \x01(\x0b\x32(.temporal.api.streamservice.v1.StreamRef"\xc7\x01\n\x16\x44\x65scribeStreamResponse\x12;\n\x06status\x18\x01 \x01(\x0e\x32+.temporal.api.streamservice.v1.StreamStatus\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nauto_close\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration*\xc8\x01\n\x0cProducerKind\x12\x1d\n\x19PRODUCER_KIND_UNSPECIFIED\x10\x00\x12\x1a\n\x16PRODUCER_KIND_WORKFLOW\x10\x01\x12 \n\x1cPRODUCER_KIND_LOCAL_ACTIVITY\x10\x02\x12\x1a\n\x16PRODUCER_KIND_ACTIVITY\x10\x03\x12\x18\n\x14PRODUCER_KIND_CLIENT\x10\x04\x12%\n!PRODUCER_KIND_STANDALONE_ACTIVITY\x10\x05*\xb3\x01\n\x10StreamRecordKind\x12"\n\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n\x17STREAM_RECORD_KIND_DATA\x10\x01\x12!\n\x1dSTREAM_RECORD_KIND_SUPERSEDED\x10\x02\x12\x1c\n\x18STREAM_RECORD_KIND_CLOSE\x10\x03"\x04\x08\x04\x10\x04*\x17STREAM_RECORD_KIND_IDLE*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03*U\n\tReadStart\x12\x1a\n\x16READ_START_UNSPECIFIED\x10\x00\x12\x18\n\x14READ_START_BEGINNING\x10\x01\x12\x12\n\x0eREAD_START_END\x10\x02\x42\xb2\x01\n io.temporal.api.streamservice.v1B\x12StreamServiceProtoP\x01Z1go.temporal.io/api/streamservice/v1;streamservice\xaa\x02\x1fTemporalio.Api.StreamService.V1\xea\x02"Temporalio::Api::StreamService::V1b\x06proto3' ) _PRODUCERKIND = DESCRIPTOR.enum_types_by_name["ProducerKind"] @@ -262,6 +262,7 @@ if _descriptor._USE_C_DESCRIPTORS == False: DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b'\n io.temporal.api.streamservice.v1B\022StreamServiceProtoP\001Z1go.temporal.io/api/streamservice/v1;streamservice\252\002\037Temporalio.Api.StreamService.V1\352\002"Temporalio::Api::StreamService::V1' _PRODUCERKIND._serialized_start = 3518 _PRODUCERKIND._serialized_end = 3718 _STREAMRECORDKIND._serialized_start = 3721 diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index 9304de8ac..ee3eed841 100644 --- a/temporalio/bridge/Cargo.lock +++ b/temporalio/bridge/Cargo.lock @@ -2879,7 +2879,7 @@ dependencies = [ [[package]] name = "temporalio-streams" version = "0.1.0" -source = "git+https://github.com/moetemp/sdk-streams?rev=2cc3eee3760a24b2feef30dc9471010cd49750e6#2cc3eee3760a24b2feef30dc9471010cd49750e6" +source = "git+https://github.com/moetemp/sdk-streams?rev=ef17adf7b4532ac652d53201f7e8703082e4bbb0#ef17adf7b4532ac652d53201f7e8703082e4bbb0" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index 93f011cfb..254746cd9 100644 --- a/temporalio/bridge/Cargo.toml +++ b/temporalio/bridge/Cargo.toml @@ -94,7 +94,7 @@ url = "2.2" # crates, so the bridge sees one type of each. [dependencies.temporalio-streams] git = "https://github.com/moetemp/sdk-streams" -rev = "2cc3eee3760a24b2feef30dc9471010cd49750e6" +rev = "ef17adf7b4532ac652d53201f7e8703082e4bbb0" default-features = false optional = true diff --git a/temporalio/contrib/streams/_activity.py b/temporalio/contrib/streams/_activity.py index 4c171d937..ad6cfe3c0 100644 --- a/temporalio/contrib/streams/_activity.py +++ b/temporalio/contrib/streams/_activity.py @@ -36,9 +36,7 @@ async def known_id() -> str: return stream_id writer = StreamWriter( - backend=self.backend, - converter=self.converter, - namespace=self.namespace, + writers=self.backend.writers(self.namespace, self.converter), stream_id=known_id, producer=self.producer, ) diff --git a/temporalio/contrib/streams/_backend.py b/temporalio/contrib/streams/_backend.py index 76740fb1e..d17682742 100644 --- a/temporalio/contrib/streams/_backend.py +++ b/temporalio/contrib/streams/_backend.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import timedelta -from typing import Protocol, TypeVar +from typing import TYPE_CHECKING, Protocol, TypeVar import google.protobuf.message @@ -15,6 +15,10 @@ from temporalio.contrib.streams.proto import v1 as ext_pb from temporalio.contrib.streams.system import client as service_client +if TYPE_CHECKING: + # The writer bindings import this package's protos, so they load after it. + from temporalio.contrib.streams.writer_system import client as writer_client + _M = TypeVar("_M", bound=google.protobuf.message.Message) _REQUESTS: dict[str, type[google.protobuf.message.Message]] = { @@ -23,6 +27,9 @@ "CloseStream": stream_pb.CloseStreamRequest, "ReadRecords": stream_pb.ReadRecordsRequest, "DescribeStream": stream_pb.DescribeStreamRequest, + "OpenWriter": ext_pb.OpenWriterRequest, + "WriteRecords": ext_pb.WriteRecordsRequest, + "CloseWriter": ext_pb.CloseWriterRequest, } _RESPONSES: dict[str, type[google.protobuf.message.Message]] = { @@ -31,6 +38,9 @@ "CloseStream": stream_pb.CloseStreamResponse, "ReadRecords": stream_pb.ReadRecordsResponse, "DescribeStream": stream_pb.DescribeStreamResponse, + "OpenWriter": ext_pb.OpenWriterResponse, + "WriteRecords": ext_pb.WriteRecordsResponse, + "CloseWriter": ext_pb.CloseWriterResponse, } @@ -45,24 +55,10 @@ def service( """The Stream service for ``namespace``.""" ... - async def open_writer(self, request: ext_pb.OpenWriterRequest) -> int: - """Open the writer of an Activity or client Producer, and return its id.""" - ... - - async def write_records( - self, writer_id: int, records: list[stream_pb.AppendRecord] - ) -> int: - """Write Records for a writer, and return the last Record's offset.""" - ... - - async def close_writer(self, writer_id: int) -> int: - """Drop a writer. The Stream stays open.""" - ... - - async def flush_run_buffer( - self, namespace: str, run_id: str, workflow_task: stream_pb.WorkflowTaskEpoch - ) -> int: - """Write a run's buffered calls before a Local Activity of the run starts.""" + def writers( + self, namespace: str, data_converter: temporalio.converter.DataConverter + ) -> writer_client.StreamWriterClient: + """The writers of Activity and client Producers for ``namespace``.""" ... @@ -90,6 +86,16 @@ def service( _ExtensionTransport(self, namespace), data_converter=data_converter ) + def writers( + self, namespace: str, data_converter: temporalio.converter.DataConverter + ) -> writer_client.StreamWriterClient: + """The generated writer client over the extension.""" + from temporalio.contrib.streams.writer_system import client + + return client.StreamWriterClient( + _ExtensionTransport(self, namespace), data_converter=data_converter + ) + def call(self, method: str, request: google.protobuf.message.Message) -> bytes: """Call a synchronous extension method.""" try: @@ -107,24 +113,6 @@ async def call_async( raise _errors.from_extension_error(err) from None return response.FromString(data) - async def open_writer(self, request: ext_pb.OpenWriterRequest) -> int: - """Open a buffered writer and return its id.""" - response = await self.call_async( - "OpenWriter", request, ext_pb.OpenWriterResponse - ) - return response.writer_id - - async def write_records( - self, writer_id: int, records: list[stream_pb.AppendRecord] - ) -> int: - """Write Records at once for a writer. Returns the last Record's offset.""" - response = await self.call_async( - "WriteRecords", - ext_pb.WriteRecordsRequest(writer_id=writer_id, records=records), - ext_pb.WriteRecordsResponse, - ) - return response.last_offset - async def flush_run_buffer( self, namespace: str, run_id: str, workflow_task: stream_pb.WorkflowTaskEpoch ) -> int: @@ -141,15 +129,6 @@ async def flush_run_buffer( ) return response.records - async def close_writer(self, writer_id: int) -> int: - """Flush and drop a writer. The Stream stays open.""" - response = await self.call_async( - "CloseWriter", - ext_pb.CloseWriterRequest(writer_id=writer_id), - ext_pb.CloseWriterResponse, - ) - return response.last_offset - async def close_writers(self) -> int: """Drop every open writer of the extension, and return how many failed.""" response = await self.call_async( @@ -183,7 +162,8 @@ async def call( ) -> temporalio.api.common.v1.Payload: request = _REQUESTS[operation].FromString(input.data) # The IDL leaves the namespace out, since the server knows it. - setattr(request, "namespace", self._namespace) + if "namespace" in request.DESCRIPTOR.fields_by_name: + setattr(request, "namespace", self._namespace) response = await self._backend.call_async( operation, request, _RESPONSES[operation] ) diff --git a/temporalio/contrib/streams/_stream_client.py b/temporalio/contrib/streams/_stream_client.py index 8ea7b681d..ac878a569 100644 --- a/temporalio/contrib/streams/_stream_client.py +++ b/temporalio/contrib/streams/_stream_client.py @@ -170,9 +170,7 @@ def _access(self) -> Access: def writer() -> StreamWriter: if self._writer is None: self._writer = StreamWriter( - backend=backend, - converter=client.data_converter, - namespace=client.namespace, + writers=backend.writers(client.namespace, client.data_converter), stream_id=self._stream_id, producer=_client_producer(client), ) diff --git a/temporalio/contrib/streams/_writer.py b/temporalio/contrib/streams/_writer.py index 11cffb501..8b7d6e833 100644 --- a/temporalio/contrib/streams/_writer.py +++ b/temporalio/contrib/streams/_writer.py @@ -4,34 +4,33 @@ import asyncio from collections.abc import Awaitable, Callable, Sequence -from typing import Any +from typing import TYPE_CHECKING, Any import temporalio.api.streamservice.v1 as stream_pb -import temporalio.converter from temporalio.contrib.streams import _errors -from temporalio.contrib.streams._backend import StreamBackend -from temporalio.contrib.streams.proto import v1 as ext_pb + +if TYPE_CHECKING: + from temporalio.contrib.streams.writer_system import client as writer_client class StreamWriter: - """Writes the Records of one Producer on one Stream, through the crate's writer.""" + """Writes the Records of one Producer on one Stream, through the generated Writer.""" def __init__( self, *, - backend: StreamBackend, - converter: temporalio.converter.DataConverter, - namespace: str, + writers: writer_client.StreamWriterClient, stream_id: Callable[[], Awaitable[str]], producer: stream_pb.Producer, ) -> None: - """Create a writer. It opens on its first write.""" - self._backend = backend - self._converter = converter - self._namespace = namespace + """Create a writer. It opens on its first write. + + ``writers`` runs the values through the user's converter and codec. + """ + self._writers = writers self._resolve_id = stream_id self._producer = producer - self._writer_id: int | None = None + self._writer: writer_client.Writer | None = None self._open_lock = asyncio.Lock() self._closed = False @@ -50,38 +49,43 @@ async def withdraw_earlier_attempts(self) -> None: except _errors.StreamNotFoundError: pass - async def open(self) -> int: + async def open(self) -> writer_client.Writer: """Open the crate's writer once. The open of a new attempt withdraws the earlier one's Records, even when it writes nothing. """ + from temporalio.contrib.streams.writer_system import models + async with self._open_lock: - if self._writer_id is None: - self._writer_id = await self._backend.open_writer( - ext_pb.OpenWriterRequest( - namespace=self._namespace, - stream=stream_pb.StreamRef(id=await self._resolve_id()), - producer=self._producer, + if self._writer is None: + producer = models._ProducerTransferTypeConverter().from_transfer_type( + self._producer, models.Producer + ) + self._writer = await self._writers.open_writer( + models.OpenWriterRequest( + stream=models.StreamRef(id=await self._resolve_id()), + producer=producer, ) ) - return self._writer_id + return self._writer async def write(self, topic: str, values: Sequence[Any]) -> int: """Write the values as Records now. Returns the last Record's offset.""" + from temporalio.contrib.streams.writer_system import models + if self._closed: raise _errors.StreamClosedError( "The writer is closed", reason="stream_closed" ) - writer_id = await self.open() - bodies = await self._converter.encode(list(values)) - return await self._backend.write_records( - writer_id, - [stream_pb.AppendRecord(topic=topic, body=body) for body in bodies], + writer = await self.open() + answer = await writer.write_records( + [models.AppendRecord(topic=topic, body=value) for value in values] ) + return answer.last_offset or 0 async def close(self) -> None: """Drop the writer. The Stream stays open.""" if self._closed: return self._closed = True - if self._writer_id is not None: - await self._backend.close_writer(self._writer_id) + if self._writer is not None: + await self._writer.close() diff --git a/temporalio/contrib/streams/redis.py b/temporalio/contrib/streams/redis.py index 58398cbe7..69733147e 100644 --- a/temporalio/contrib/streams/redis.py +++ b/temporalio/contrib/streams/redis.py @@ -606,7 +606,7 @@ async def execute_activity( streams = _activity_streams(input, self._plugin) token = _activity.current.set(streams) try: - await _flush_run_before_local_activity(streams) + await _flush_run_before_local_activity(self._plugin._backend, streams) if temporalio.activity.info().attempt > 1: # The open of a new attempt withdraws the Records of the earlier ones, also # when this attempt publishes nothing. It creates no Stream. @@ -624,7 +624,9 @@ async def execute_activity( _activity.current.reset(token) -async def _flush_run_before_local_activity(streams: _activity.ActivityStreams) -> None: +async def _flush_run_before_local_activity( + backend: _ExtensionStreamBackend, streams: _activity.ActivityStreams +) -> None: # The run's buffered Records go before the Local Activity's, in the epoch of the Workflow # Task that dispatched it. A failure fails the attempt, and the run's next completion # fails too, so its retry withdraws what was written. @@ -637,9 +639,7 @@ async def _flush_run_before_local_activity(streams: _activity.ActivityStreams) - ) if task.started_time is not None: epoch.started_time.FromDatetime(task.started_time) - await streams.backend.flush_run_buffer( - streams.namespace, info.workflow_run_id, epoch - ) + await backend.flush_run_buffer(streams.namespace, info.workflow_run_id, epoch) def _stream_arguments(args: Sequence[Any]) -> list[StreamHandle[Any]]: diff --git a/temporalio/contrib/streams/system/_resources/__init__.py b/temporalio/contrib/streams/system/_resources/__init__.py index 66de98b6e..4fa8f0ac2 100644 --- a/temporalio/contrib/streams/system/_resources/__init__.py +++ b/temporalio/contrib/streams/system/_resources/__init__.py @@ -2,8 +2,12 @@ from __future__ import annotations -from .stream import Stream +from .stream import ( + CreateStreamResult, + Stream, +) __all__ = [ "Stream", + "CreateStreamResult", ] diff --git a/temporalio/contrib/streams/system/_resources/stream.py b/temporalio/contrib/streams/system/_resources/stream.py index 457699b68..184b846fd 100644 --- a/temporalio/contrib/streams/system/_resources/stream.py +++ b/temporalio/contrib/streams/system/_resources/stream.py @@ -174,6 +174,15 @@ async def describe( return await handle +@dataclasses.dataclass(slots=True) +class CreateStreamResult: + stream: Stream + created: bool | None = None + status: ( + temporalio.api.streamservice.v1.stream_service_pb2.StreamStatus.ValueType | None + ) = None + + async def _append_records( request: AppendRecordsRequest, ) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]: diff --git a/temporalio/contrib/streams/system/models.py b/temporalio/contrib/streams/system/models.py index 8d19543f7..2304fc581 100644 --- a/temporalio/contrib/streams/system/models.py +++ b/temporalio/contrib/streams/system/models.py @@ -13,10 +13,6 @@ import temporalio.converter import temporalio.nexus.system._support as _support -if typing.TYPE_CHECKING: - from ._resources import Stream - - StreamOffset = typing.NewType("StreamOffset", int) """A position in an ordered log, as the service gives it. @@ -1252,12 +1248,3 @@ def to_transfer_type( temporalio.converter.transfer_type_convertible( _DescribeStreamResponseTransferTypeConverter )(DescribeStreamResponse) # pyright: ignore[reportUnusedCallResult] - - -@dataclasses.dataclass(slots=True) -class CreateStreamResult: - stream: Stream - created: bool | None = None - status: ( - temporalio.api.streamservice.v1.stream_service_pb2.StreamStatus.ValueType | None - ) = None diff --git a/temporalio/contrib/streams/system/operations/create_stream.py b/temporalio/contrib/streams/system/operations/create_stream.py index ad7dd041b..39efa09da 100644 --- a/temporalio/contrib/streams/system/operations/create_stream.py +++ b/temporalio/contrib/streams/system/operations/create_stream.py @@ -6,11 +6,13 @@ import temporalio.workflow -from .._resources import Stream +from .._resources import ( + CreateStreamResult, + Stream, +) from ..models import ( CreateStreamRequest, CreateStreamResponse, - CreateStreamResult, ) diff --git a/temporalio/contrib/streams/writer_system/__init__.py b/temporalio/contrib/streams/writer_system/__init__.py new file mode 100644 index 000000000..48a201704 --- /dev/null +++ b/temporalio/contrib/streams/writer_system/__init__.py @@ -0,0 +1,5 @@ +# Generated by nexgen v0.2.9. DO NOT EDIT! + +from __future__ import annotations + +__all__ = [] diff --git a/temporalio/contrib/streams/writer_system/client.py b/temporalio/contrib/streams/writer_system/client.py new file mode 100644 index 000000000..4d94097fb --- /dev/null +++ b/temporalio/contrib/streams/writer_system/client.py @@ -0,0 +1,288 @@ +# Generated by nexgen v0.2.9. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import contextlib +import typing + +import google.protobuf.json_format +import google.protobuf.message + +import temporalio.api.common.v1 +import temporalio.converter +import temporalio.nexus.system + +from .models import ( + AppendRecord, + CloseWriterRequest, + CloseWriterResponse, + CloseWritersRequest, + CloseWritersResponse, + OpenWriterRequest, + WriteRecordsRequest, + WriteRecordsResponse, + WriterResult, + _CloseWriterRequestTransferTypeConverter, + _CloseWriterResponseTransferTypeConverter, + _CloseWritersRequestTransferTypeConverter, + _CloseWritersResponseTransferTypeConverter, + _OpenWriterRequestTransferTypeConverter, + _WriteRecordsRequestTransferTypeConverter, + _WriteRecordsResponseTransferTypeConverter, + _WriterResultTransferTypeConverter, +) + + +class Transport(typing.Protocol): + """Carries one operation call to a service and its answer back. + + A transport moves payloads only. The caller converts its models to and + from payloads. A proto-backed model travels as a ``binary/protobuf`` + payload, and an answer for one can be ``binary/protobuf``, + ``json/protobuf`` or ``json/plain``. + """ + + async def call( + self, + *, + service: str, + operation: str, + input: temporalio.api.common.v1.Payload, + ) -> temporalio.api.common.v1.Payload: + """Send one call and return its answer. + + Raise :py:class:`TransportError` for a call that failed. + """ + ... + + +class TransportError(Exception): + """A call that the transport could not complete. + + ``retryable`` says whether the same call can succeed later. + """ + + def __init__(self, message: str, *, retryable: bool = False) -> None: + """Create a transport error.""" + super().__init__(message) + self.retryable: bool = retryable + + +_MessageT = typing.TypeVar("_MessageT", bound=google.protobuf.message.Message) + + +def _proto_payload( + message: google.protobuf.message.Message, +) -> temporalio.api.common.v1.Payload: + return temporalio.api.common.v1.Payload( + metadata={ + "encoding": b"binary/protobuf", + "messageType": message.DESCRIPTOR.full_name.encode(), + }, + data=message.SerializeToString(), + ) + + +def _proto_message( + payload: temporalio.api.common.v1.Payload, + message_type: type[_MessageT] | None, +) -> _MessageT: + if message_type is None: + raise TypeError("the answer model has no proto message type") + message = message_type() + encoding = payload.metadata.get("encoding", b"") + if encoding == b"binary/protobuf": + _ = message.ParseFromString(payload.data) + elif encoding in (b"json/protobuf", b"json/plain"): + _ = google.protobuf.json_format.Parse(payload.data, message) + elif encoding != b"binary/null": + raise ValueError( + f"cannot read {message.DESCRIPTOR.full_name} from a {encoding.decode()!r} payload" + ) + return message + + +# The codec runs on the user payloads inside a proto-backed model, Record bodies and +# failure details included, and never on the model's own payload, which the service reads. +def _payloads_in( + message: google.protobuf.message.Message, +) -> list[temporalio.api.common.v1.Payload]: + if isinstance(message, temporalio.api.common.v1.Payload): + return [message] + found: list[temporalio.api.common.v1.Payload] = [] + for _, value in message.ListFields(): + if isinstance(value, google.protobuf.message.Message): + found.extend(_payloads_in(value)) + elif not isinstance(value, (str, bytes, int, float, bool)): + items = typing.cast( + "collections.abc.Iterable[object]", + value.values() if hasattr(value, "values") else value, + ) + for item in items: + if isinstance(item, google.protobuf.message.Message): + found.extend(_payloads_in(item)) + return found + + +async def _encode_payloads( + message: google.protobuf.message.Message, + codec: temporalio.converter.PayloadCodec | None, +) -> None: + if codec is None: + return + for payload in _payloads_in(message): + [encoded] = await codec.encode([payload]) + payload.CopyFrom(encoded) + + +async def _decode_payloads( + message: google.protobuf.message.Message, + codec: temporalio.converter.PayloadCodec | None, +) -> None: + if codec is None: + return + for payload in _payloads_in(message): + [decoded] = await codec.decode([payload]) + payload.CopyFrom(decoded) + + +class StreamWriterClient: + """Calls the `temporal.sdk.streams.v1.StreamWriter` operations over a transport. + + For code outside a Workflow. Workflow code uses the operation stubs. + """ + + def __init__( + self, + transport: Transport, + *, + data_converter: temporalio.converter.DataConverter | None = None, + ) -> None: + """Create a caller. + + Args: + transport: Carries each call to the service. + data_converter: Converts and encodes the user values. A proto-backed + model travels as is, and each payload and failure inside it goes + through the converter and the codec, as in Workflow code. A model + that is not proto-backed is one user value. Defaults to the SDK's + default data converter. + """ + self._transport = transport + self._data_converter = ( + data_converter or temporalio.converter.DataConverter.default + ) + + def _converters(self) -> contextlib.AbstractContextManager[None]: + return temporalio.nexus.system._user_converter_context( + temporalio.nexus.system._SystemNexusUserConverters( + payload_converter=self._data_converter.payload_converter, + failure_converter=self._data_converter.failure_converter, + ) + ) + + async def _call( + self, operation: str, payload: temporalio.api.common.v1.Payload + ) -> temporalio.api.common.v1.Payload: + return await self._transport.call( + service="temporal.sdk.streams.v1.StreamWriter", + operation=operation, + input=payload, + ) + + async def open_writer(self, request: OpenWriterRequest) -> Writer: + with self._converters(): + message = _OpenWriterRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("OpenWriter", payload) + converter = _WriterResultTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + result = converter.from_transfer_type(message, WriterResult) + if result.writer_id is None: + raise RuntimeError("OpenWriter answered without `writer_id`") + return Writer( + self, + writer_id=result.writer_id, + ) + + async def write_records(self, request: WriteRecordsRequest) -> WriteRecordsResponse: + with self._converters(): + message = _WriteRecordsRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("WriteRecords", payload) + converter = _WriteRecordsResponseTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + return converter.from_transfer_type(message, WriteRecordsResponse) + + async def close_writer(self, request: CloseWriterRequest) -> CloseWriterResponse: + with self._converters(): + message = _CloseWriterRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("CloseWriter", payload) + converter = _CloseWriterResponseTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + return converter.from_transfer_type(message, CloseWriterResponse) + + async def close_writers(self, request: CloseWritersRequest) -> CloseWritersResponse: + """Drop every open writer, for example at shutdown.""" + with self._converters(): + message = _CloseWritersRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("CloseWriters", payload) + converter = _CloseWritersResponseTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + return converter.from_transfer_type(message, CloseWritersResponse) + + +class Writer: + """An open writer of one Producer. The crate keeps its epoch and sequence.""" + + def __init__( + self, + client: StreamWriterClient, + writer_id: int, + ) -> None: + """Bind the handle to a caller.""" + self._client = client + self.writer_id = writer_id + + async def write_records( + self, + records: list[AppendRecord] | None = None, + ) -> WriteRecordsResponse: + """Write Records at once. The answer has the first and the last offset.""" + request = WriteRecordsRequest( + writer_id=self.writer_id, + records=records or [], + ) + return await self._client.write_records(request) + + async def close( + self, + ) -> CloseWriterResponse: + """Wait for a write in flight, then drop the writer. The Stream stays open.""" + request = CloseWriterRequest( + writer_id=self.writer_id, + ) + return await self._client.close_writer(request) diff --git a/temporalio/contrib/streams/writer_system/models.py b/temporalio/contrib/streams/writer_system/models.py new file mode 100644 index 000000000..8a51a6493 --- /dev/null +++ b/temporalio/contrib/streams/writer_system/models.py @@ -0,0 +1,699 @@ +# Generated by nexgen v0.2.9. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses +import datetime +import typing + +import google.protobuf.timestamp_pb2 +import typing_extensions + +import temporalio.api.streamservice.v1.stream_service_pb2 +import temporalio.contrib.streams.proto.v1.extension_pb2 +import temporalio.converter +import temporalio.nexus.system._support as _support + + +@dataclasses.dataclass(slots=True) +class OpenWriterRequest: + """Open a writer for one Producer of one Stream.""" + + stream: StreamRef | None = None + producer: Producer | None = None + max_batch_size: int | None = None + max_retry_duration: datetime.timedelta | None = None + + +class _OpenWriterRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OpenWriterRequest, + temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterRequest, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterRequest] | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterRequest, + type_hint: type[OpenWriterRequest], + ) -> OpenWriterRequest: + return OpenWriterRequest( + stream=_StreamRefTransferTypeConverter().from_transfer_type( + value.stream, StreamRef + ) + if value.HasField("stream") + else None, + producer=_ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + if value.HasField("producer") + else None, + max_batch_size=value.max_batch_size if value.max_batch_size != 0 else None, + max_retry_duration=_support.duration_from_proto(value.max_retry_duration) + if value.HasField("max_retry_duration") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: OpenWriterRequest, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterRequest: + message = temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterRequest() + if value.stream is not None: + message.stream.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream) + ) + if value.producer is not None: + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + if value.max_batch_size is not None: + message.max_batch_size = value.max_batch_size + if value.max_retry_duration is not None: + message.max_retry_duration.CopyFrom( + _support.duration_to_proto(value.max_retry_duration) + ) + return message + + +temporalio.converter.transfer_type_convertible(_OpenWriterRequestTransferTypeConverter)( + OpenWriterRequest +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class StreamRef: + """Identifies one Stream. The id is opaque to callers.""" + + id: str + workflow_id: str | None = None + first_execution_run_id: str | None = None + backend: str | None = None + + +class _StreamRefTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamRef, temporalio.api.streamservice.v1.stream_service_pb2.StreamRef + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.stream_service_pb2.StreamRef] | None + ) = temporalio.api.streamservice.v1.stream_service_pb2.StreamRef + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.stream_service_pb2.StreamRef, + type_hint: type[StreamRef], + ) -> StreamRef: + if not value.id: + raise ValueError("missing required field StreamRef.id") + id = value.id + return StreamRef( + id=id, + workflow_id=value.workflow_id if bool(value.workflow_id) else None, + first_execution_run_id=value.first_execution_run_id + if bool(value.first_execution_run_id) + else None, + backend=value.backend if bool(value.backend) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamRef, + ) -> temporalio.api.streamservice.v1.stream_service_pb2.StreamRef: + message = temporalio.api.streamservice.v1.stream_service_pb2.StreamRef() + message.id = value.id + if value.workflow_id is not None: + message.workflow_id = value.workflow_id + if value.first_execution_run_id is not None: + message.first_execution_run_id = value.first_execution_run_id + if value.backend is not None: + message.backend = value.backend + return message + + +temporalio.converter.transfer_type_convertible(_StreamRefTransferTypeConverter)( + StreamRef +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True, kw_only=True) +class Producer: + """The Producer of a batch.""" + + kind: ( + temporalio.api.streamservice.v1.stream_service_pb2.ProducerKind.ValueType | None + ) = None + id: str + epoch: ProducerEpoch | None = None + run_id: str | None = None + + +class _ProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Producer, temporalio.api.streamservice.v1.stream_service_pb2.Producer + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.stream_service_pb2.Producer] | None + ) = temporalio.api.streamservice.v1.stream_service_pb2.Producer + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.stream_service_pb2.Producer, + type_hint: type[Producer], + ) -> Producer: + if not value.id: + raise ValueError("missing required field Producer.id") + id = value.id + return Producer( + kind=temporalio.api.streamservice.v1.stream_service_pb2.ProducerKind.ValueType( + value.kind + ) + if value.kind != 0 + else None, + id=id, + epoch=_ProducerEpochTransferTypeConverter().from_transfer_type( + value.epoch, ProducerEpoch + ) + if value.HasField("epoch") + else None, + run_id=value.run_id if bool(value.run_id) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Producer, + ) -> temporalio.api.streamservice.v1.stream_service_pb2.Producer: + message = temporalio.api.streamservice.v1.stream_service_pb2.Producer() + if value.kind is not None: + message.kind = temporalio.api.streamservice.v1.stream_service_pb2.ProducerKind.ValueType( + value.kind + ) + message.id = value.id + if value.epoch is not None: + message.epoch.CopyFrom( + _ProducerEpochTransferTypeConverter().to_transfer_type(value.epoch) + ) + if value.run_id is not None: + message.run_id = value.run_id + return message + + +temporalio.converter.transfer_type_convertible(_ProducerTransferTypeConverter)(Producer) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ProducerEpoch: + """The ordered attempt of a Producer. A higher epoch supersedes a lower one.""" + + task_started_event_id: int | None = None + task_attempt: int | None = None + task_started_time: Timestamp | None = None + attempt_scheduled_time: Timestamp | None = None + attempt: int | None = None + + +class _ProducerEpochTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ProducerEpoch, temporalio.api.streamservice.v1.stream_service_pb2.ProducerEpoch + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.stream_service_pb2.ProducerEpoch] | None + ) = temporalio.api.streamservice.v1.stream_service_pb2.ProducerEpoch + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.stream_service_pb2.ProducerEpoch, + type_hint: type[ProducerEpoch], + ) -> ProducerEpoch: + return ProducerEpoch( + task_started_event_id=value.task_started_event_id + if value.task_started_event_id != 0 + else None, + task_attempt=value.task_attempt if value.task_attempt != 0 else None, + task_started_time=_TimestampTransferTypeConverter().from_transfer_type( + value.task_started_time, Timestamp + ) + if value.HasField("task_started_time") + else None, + attempt_scheduled_time=_TimestampTransferTypeConverter().from_transfer_type( + value.attempt_scheduled_time, Timestamp + ) + if value.HasField("attempt_scheduled_time") + else None, + attempt=value.attempt if value.attempt != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ProducerEpoch, + ) -> temporalio.api.streamservice.v1.stream_service_pb2.ProducerEpoch: + message = temporalio.api.streamservice.v1.stream_service_pb2.ProducerEpoch() + if value.task_started_event_id is not None: + message.task_started_event_id = value.task_started_event_id + if value.task_attempt is not None: + message.task_attempt = value.task_attempt + if value.task_started_time is not None: + message.task_started_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type( + value.task_started_time + ) + ) + if value.attempt_scheduled_time is not None: + message.attempt_scheduled_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type( + value.attempt_scheduled_time + ) + ) + if value.attempt is not None: + message.attempt = value.attempt + return message + + +temporalio.converter.transfer_type_convertible(_ProducerEpochTransferTypeConverter)( + ProducerEpoch +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Timestamp: + seconds: int | None = None + nanos: int | None = None + + +class _TimestampTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Timestamp, google.protobuf.timestamp_pb2.Timestamp + ] +): + transfer_type: type[google.protobuf.timestamp_pb2.Timestamp] | None = ( + google.protobuf.timestamp_pb2.Timestamp + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: google.protobuf.timestamp_pb2.Timestamp, + type_hint: type[Timestamp], + ) -> Timestamp: + return Timestamp( + seconds=value.seconds if value.seconds != 0 else None, + nanos=value.nanos if value.nanos != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Timestamp, + ) -> google.protobuf.timestamp_pb2.Timestamp: + message = google.protobuf.timestamp_pb2.Timestamp() + if value.seconds is not None: + message.seconds = value.seconds + if value.nanos is not None: + message.nanos = value.nanos + return message + + +temporalio.converter.transfer_type_convertible(_TimestampTransferTypeConverter)( + Timestamp +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class WriterResult: + writer_id: int | None = None + + +class _WriterResultTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + WriterResult, + temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterResponse, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterResponse] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterResponse, + type_hint: type[WriterResult], + ) -> WriterResult: + return WriterResult( + writer_id=value.writer_id if value.writer_id != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: WriterResult, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterResponse: + message = temporalio.contrib.streams.proto.v1.extension_pb2.OpenWriterResponse() + if value.writer_id is not None: + message.writer_id = value.writer_id + return message + + +temporalio.converter.transfer_type_convertible(_WriterResultTransferTypeConverter)( + WriterResult +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class WriteRecordsRequest: + writer_id: int | None = None + records: list[AppendRecord] | None = dataclasses.field(default_factory=list) + + +class _WriteRecordsRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + WriteRecordsRequest, + temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsRequest, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsRequest] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsRequest, + type_hint: type[WriteRecordsRequest], + ) -> WriteRecordsRequest: + return WriteRecordsRequest( + writer_id=value.writer_id if value.writer_id != 0 else None, + records=[ + _AppendRecordTransferTypeConverter().from_transfer_type( + item, AppendRecord + ) + for item in value.records + ], + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: WriteRecordsRequest, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsRequest: + message = ( + temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsRequest() + ) + if value.writer_id is not None: + message.writer_id = value.writer_id + if value.records: + for element in value.records: + item = message.records.add() + item.CopyFrom( + _AppendRecordTransferTypeConverter().to_transfer_type(element) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _WriteRecordsRequestTransferTypeConverter +)(WriteRecordsRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecord: + """One Record to append.""" + + topic: str + body: typing.Any + + +class _AppendRecordTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecord, temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord] | None + ) = temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord, + type_hint: type[AppendRecord], + ) -> AppendRecord: + if not value.topic: + raise ValueError("missing required field AppendRecord.topic") + topic = value.topic + if not value.HasField("body"): + raise ValueError("missing required field AppendRecord.body") + body = _support.payload_from_proto(value.body) + return AppendRecord( + topic=topic, + body=body, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecord, + ) -> temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord: + message = temporalio.api.streamservice.v1.stream_service_pb2.AppendRecord() + message.topic = value.topic + message.body.CopyFrom(_support.payload_to_proto(value.body)) + return message + + +temporalio.converter.transfer_type_convertible(_AppendRecordTransferTypeConverter)( + AppendRecord +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class WriteRecordsResponse: + first_offset: int | None = None + last_offset: int | None = None + + +class _WriteRecordsResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + WriteRecordsResponse, + temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsResponse, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsResponse] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsResponse, + type_hint: type[WriteRecordsResponse], + ) -> WriteRecordsResponse: + return WriteRecordsResponse( + first_offset=value.first_offset if value.first_offset != 0 else None, + last_offset=value.last_offset if value.last_offset != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: WriteRecordsResponse, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsResponse: + message = ( + temporalio.contrib.streams.proto.v1.extension_pb2.WriteRecordsResponse() + ) + if value.first_offset is not None: + message.first_offset = value.first_offset + if value.last_offset is not None: + message.last_offset = value.last_offset + return message + + +temporalio.converter.transfer_type_convertible( + _WriteRecordsResponseTransferTypeConverter +)(WriteRecordsResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseWriterRequest: + writer_id: int | None = None + + +class _CloseWriterRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseWriterRequest, + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterRequest, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterRequest] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterRequest, + type_hint: type[CloseWriterRequest], + ) -> CloseWriterRequest: + return CloseWriterRequest( + writer_id=value.writer_id if value.writer_id != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseWriterRequest, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterRequest: + message = temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterRequest() + if value.writer_id is not None: + message.writer_id = value.writer_id + return message + + +temporalio.converter.transfer_type_convertible( + _CloseWriterRequestTransferTypeConverter +)(CloseWriterRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseWriterResponse: + last_offset: int | None = None + + +class _CloseWriterResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseWriterResponse, + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterResponse, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterResponse] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterResponse, + type_hint: type[CloseWriterResponse], + ) -> CloseWriterResponse: + return CloseWriterResponse( + last_offset=value.last_offset if value.last_offset != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseWriterResponse, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterResponse: + message = ( + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWriterResponse() + ) + if value.last_offset is not None: + message.last_offset = value.last_offset + return message + + +temporalio.converter.transfer_type_convertible( + _CloseWriterResponseTransferTypeConverter +)(CloseWriterResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseWritersRequest: + pass + + +class _CloseWritersRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseWritersRequest, + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersRequest, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersRequest] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersRequest, + type_hint: type[CloseWritersRequest], + ) -> CloseWritersRequest: + return CloseWritersRequest() + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseWritersRequest, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersRequest: + message = ( + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersRequest() + ) + return message + + +temporalio.converter.transfer_type_convertible( + _CloseWritersRequestTransferTypeConverter +)(CloseWritersRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseWritersResponse: + failed: int | None = None + + +class _CloseWritersResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseWritersResponse, + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersResponse, + ] +): + transfer_type: ( + type[temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersResponse] + | None + ) = temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersResponse, + type_hint: type[CloseWritersResponse], + ) -> CloseWritersResponse: + return CloseWritersResponse( + failed=value.failed if value.failed != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseWritersResponse, + ) -> temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersResponse: + message = ( + temporalio.contrib.streams.proto.v1.extension_pb2.CloseWritersResponse() + ) + if value.failed is not None: + message.failed = value.failed + return message + + +temporalio.converter.transfer_type_convertible( + _CloseWritersResponseTransferTypeConverter +)(CloseWritersResponse) # pyright: ignore[reportUnusedCallResult]