diff --git a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto index 6a42508b8..cf338ca55 100644 --- a/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto +++ b/scripts/_proto/temporal/api/streamservice/v1/stream_service.proto @@ -179,6 +179,9 @@ message ReadRecordsRequest { // End the read with an IDLE Record when no new Record comes for this long. Empty never // ends on idle. The facade counts it across reads, since one read waits at most 60 s. google.protobuf.Duration idle_timeout = 9; + // Also read the Topics that start with one of these. A Record passes when its Topic is in + // topics or starts with a prefix. Both empty reads every Topic. + repeated string topic_prefixes = 10; } message ReadRecordsResponse { diff --git a/scripts/_wit/stream-service.wit b/scripts/_wit/stream-service.wit index 65cac791c..cb29651b3 100644 --- a/scripts/_wit/stream-service.wit +++ b/scripts/_wit/stream-service.wit @@ -243,8 +243,10 @@ interface stream-service { from-offset: option, /// @nexus.doc "Where to start when from-offset is empty. Unspecified is the beginning." start: option, - /// @nexus.doc "Empty reads every Topic." + /// @nexus.doc "Topics to read, by exact match. With topic-prefixes, empty reads every Topic." topics: list, + /// @nexus.doc "Also read the Topics that start with one of these." + topic-prefixes: list, /// @nexus.doc "At most this many Records. 1 to 1000." max-records: option, /// @nexus.doc "How long to wait for a new Record before an empty answer. At most 60 s." diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.py b/temporalio/api/streamservice/v1/stream_service_pb2.py index 093f92e06..cd9592760 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"\x17\n\tStreamRef\x12\n\n\x02id\x18\x01 \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"\x7f\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"\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"\xde\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/\n\x0cidle_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration"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"\x98\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*\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*\xb1\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\x12\x1b\n\x17STREAM_RECORD_KIND_IDLE\x10\x04*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"\x17\n\tStreamRef\x12\n\n\x02id\x18\x01 \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"\x7f\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"\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"\xf6\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/\n\x0cidle_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x16\n\x0etopic_prefixes\x18\n \x03(\t"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"\x98\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*\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*\xb1\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\x12\x1b\n\x17STREAM_RECORD_KIND_IDLE\x10\x04*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' ) _PRODUCERKIND = DESCRIPTOR.enum_types_by_name["ProducerKind"] @@ -263,14 +263,14 @@ if _descriptor._USE_C_DESCRIPTORS == False: DESCRIPTOR._options = None - _PRODUCERKIND._serialized_start = 3358 - _PRODUCERKIND._serialized_end = 3558 - _STREAMRECORDKIND._serialized_start = 3561 - _STREAMRECORDKIND._serialized_end = 3738 - _STREAMSTATUS._serialized_start = 3740 - _STREAMSTATUS._serialized_end = 3862 - _READSTART._serialized_start = 3864 - _READSTART._serialized_end = 3949 + _PRODUCERKIND._serialized_start = 3382 + _PRODUCERKIND._serialized_end = 3582 + _STREAMRECORDKIND._serialized_start = 3585 + _STREAMRECORDKIND._serialized_end = 3762 + _STREAMSTATUS._serialized_start = 3764 + _STREAMSTATUS._serialized_end = 3886 + _READSTART._serialized_start = 3888 + _READSTART._serialized_end = 3973 _STREAMREF._serialized_start = 227 _STREAMREF._serialized_end = 250 _PRODUCEREPOCH._serialized_start = 253 @@ -298,11 +298,11 @@ _CLOSESTREAMRESPONSE._serialized_start = 2604 _CLOSESTREAMRESPONSE._serialized_end = 2625 _READRECORDSREQUEST._serialized_start = 2628 - _READRECORDSREQUEST._serialized_end = 2978 - _READRECORDSRESPONSE._serialized_start = 2980 - _READRECORDSRESPONSE._serialized_end = 3098 - _DESCRIBESTREAMREQUEST._serialized_start = 3100 - _DESCRIBESTREAMREQUEST._serialized_end = 3200 - _DESCRIBESTREAMRESPONSE._serialized_start = 3203 - _DESCRIBESTREAMRESPONSE._serialized_end = 3355 + _READRECORDSREQUEST._serialized_end = 3002 + _READRECORDSRESPONSE._serialized_start = 3004 + _READRECORDSRESPONSE._serialized_end = 3122 + _DESCRIBESTREAMREQUEST._serialized_start = 3124 + _DESCRIBESTREAMREQUEST._serialized_end = 3224 + _DESCRIBESTREAMRESPONSE._serialized_start = 3227 + _DESCRIBESTREAMRESPONSE._serialized_end = 3379 # @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/streamservice/v1/stream_service_pb2.pyi b/temporalio/api/streamservice/v1/stream_service_pb2.pyi index 0b35a15fa..e1f4a62bc 100644 --- a/temporalio/api/streamservice/v1/stream_service_pb2.pyi +++ b/temporalio/api/streamservice/v1/stream_service_pb2.pyi @@ -730,6 +730,7 @@ class ReadRecordsRequest(google.protobuf.message.Message): WAIT_FIELD_NUMBER: builtins.int CREATE_WAIT_FIELD_NUMBER: builtins.int IDLE_TIMEOUT_FIELD_NUMBER: builtins.int + TOPIC_PREFIXES_FIELD_NUMBER: builtins.int namespace: builtins.str @property def stream(self) -> global___StreamRef: ... @@ -754,6 +755,13 @@ class ReadRecordsRequest(google.protobuf.message.Message): """End the read with an IDLE Record when no new Record comes for this long. Empty never ends on idle. The facade counts it across reads, since one read waits at most 60 s. """ + @property + def topic_prefixes( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: + """Also read the Topics that start with one of these. A Record passes when its Topic is in + topics or starts with a prefix. Both empty reads every Topic. + """ def __init__( self, *, @@ -766,6 +774,7 @@ class ReadRecordsRequest(google.protobuf.message.Message): wait: google.protobuf.duration_pb2.Duration | None = ..., create_wait: google.protobuf.duration_pb2.Duration | None = ..., idle_timeout: google.protobuf.duration_pb2.Duration | None = ..., + topic_prefixes: collections.abc.Iterable[builtins.str] | None = ..., ) -> None: ... def HasField( self, @@ -797,6 +806,8 @@ class ReadRecordsRequest(google.protobuf.message.Message): b"start", "stream", b"stream", + "topic_prefixes", + b"topic_prefixes", "topics", b"topics", "wait", diff --git a/temporalio/bridge/Cargo.lock b/temporalio/bridge/Cargo.lock index f39ddca90..347599f2b 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=0981f62b1454e6501e3173d2115d94cb25eb9036#0981f62b1454e6501e3173d2115d94cb25eb9036" +source = "git+https://github.com/moetemp/sdk-streams?rev=1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c#1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c" dependencies = [ "async-trait", "aws-lc-rs", diff --git a/temporalio/bridge/Cargo.toml b/temporalio/bridge/Cargo.toml index 29392c3f4..ad770a3f8 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 = "0981f62b1454e6501e3173d2115d94cb25eb9036" +rev = "1c8c01f5ce87ca4ebb3fba155430920ff0a95c5c" default-features = false optional = true diff --git a/temporalio/contrib/streams/_outside.py b/temporalio/contrib/streams/_outside.py index c2454e7c8..9d663ae5d 100644 --- a/temporalio/contrib/streams/_outside.py +++ b/temporalio/contrib/streams/_outside.py @@ -99,6 +99,7 @@ def subscribe( self, *, topics: Sequence[Topic[Any]] | None = None, + topic_prefixes: Sequence[str] = (), from_offset: int | None = None, wait_new_records: bool = True, idle_timeout: timedelta | None = None, @@ -111,7 +112,9 @@ def subscribe( a failed attempt, and it passes every Topic filter. Args: - topics: Read only these Topics. Defaults to every Topic. + topics: Read these Topics, by exact match. + topic_prefixes: Also read the Topics that start with one of these. With no + ``topics`` and no prefixes, the read gives every Topic. from_offset: Start at this offset, inclusive. Defaults to the first Record. wait_new_records: Wait for new Records at the end of the Stream. idle_timeout: End with an ``Idle`` item when no new Record comes for this long. @@ -120,6 +123,7 @@ def subscribe( """ return self._read( topics=[topic.name for topic in topics] if topics else None, + topic_prefixes=topic_prefixes, types={ **self._types, **{topic.name: topic.type for topic in topics or []}, @@ -134,6 +138,7 @@ async def _read( self, *, topics: list[str] | None, + topic_prefixes: Sequence[str], types: dict[str, Any], default_type: Any, from_offset: int | None, @@ -145,6 +150,7 @@ async def _read( self._converter, await self._stream_id(), topics=topics, + topic_prefixes=topic_prefixes, types=types, default_type=default_type, from_offset=from_offset, diff --git a/temporalio/contrib/streams/_reader.py b/temporalio/contrib/streams/_reader.py index 05ad28df4..2be35ce74 100644 --- a/temporalio/contrib/streams/_reader.py +++ b/temporalio/contrib/streams/_reader.py @@ -30,6 +30,7 @@ async def read( stream_id: str, *, topics: Sequence[str] | None, + topic_prefixes: Sequence[str] = (), types: Mapping[str, Any], default_type: Any = None, from_offset: int | None, @@ -61,6 +62,7 @@ async def read( from_offset=offset, start=None if offset is not None else ReadStart.READ_START_BEGINNING, topics=names, + topic_prefixes=list(topic_prefixes), idle_timeout=idle_timeout, create_wait=min(create_left, create_step) if create_left else None, ) diff --git a/temporalio/contrib/streams/_stream.py b/temporalio/contrib/streams/_stream.py index 6d426d9b5..16a275792 100644 --- a/temporalio/contrib/streams/_stream.py +++ b/temporalio/contrib/streams/_stream.py @@ -39,6 +39,7 @@ def subscribe( self, *, topics: Sequence[Topic[Any]] | None = None, + topic_prefixes: Sequence[str] = (), from_offset: int | None = None, wait_new_records: bool = True, idle_timeout: timedelta | None = None, diff --git a/temporalio/contrib/streams/system/client.py b/temporalio/contrib/streams/system/client.py index 436ed9553..d76a8c93c 100644 --- a/temporalio/contrib/streams/system/client.py +++ b/temporalio/contrib/streams/system/client.py @@ -511,6 +511,7 @@ async def read_records( start: temporalio.api.streamservice.v1.stream_service_pb2.ReadStart.ValueType | None = None, topics: list[str] | None = None, + topic_prefixes: list[str] | None = None, max_records: int | None = None, wait: datetime.timedelta | None = None, create_wait: datetime.timedelta | None = None, @@ -522,6 +523,7 @@ async def read_records( from_offset=from_offset, start=start, topics=topics or [], + topic_prefixes=topic_prefixes or [], max_records=max_records, wait=wait, create_wait=create_wait, @@ -536,6 +538,7 @@ async def read_records_until_records( start: temporalio.api.streamservice.v1.stream_service_pb2.ReadStart.ValueType | None = None, topics: list[str] | None = None, + topic_prefixes: list[str] | None = None, max_records: int | None = None, create_wait: datetime.timedelta | None = None, idle_timeout: datetime.timedelta | None = None, @@ -547,6 +550,7 @@ async def read_records_until_records( from_offset=from_offset, start=start, topics=topics or [], + topic_prefixes=topic_prefixes or [], max_records=max_records, create_wait=create_wait, idle_timeout=idle_timeout, diff --git a/temporalio/contrib/streams/system/models.py b/temporalio/contrib/streams/system/models.py index 244759ba4..54bcc87ae 100644 --- a/temporalio/contrib/streams/system/models.py +++ b/temporalio/contrib/streams/system/models.py @@ -778,6 +778,7 @@ class ReadRecordsRequest: temporalio.api.streamservice.v1.stream_service_pb2.ReadStart.ValueType | None ) = None topics: list[str] | None = dataclasses.field(default_factory=list) + topic_prefixes: list[str] | None = dataclasses.field(default_factory=list) max_records: int | None = None wait: datetime.timedelta | None = None create_wait: datetime.timedelta | None = None @@ -814,6 +815,7 @@ def from_transfer_type( if value.start != 0 else None, topics=list(value.topics), + topic_prefixes=list(value.topic_prefixes), max_records=value.max_records if value.max_records != 0 else None, wait=_support.duration_from_proto(value.wait) if value.HasField("wait") @@ -848,6 +850,8 @@ def to_transfer_type( ) if value.topics: message.topics.extend(value.topics) + if value.topic_prefixes: + message.topic_prefixes.extend(value.topic_prefixes) if value.max_records is not None: message.max_records = value.max_records if value.wait is not None: diff --git a/temporalio/contrib/streams/system/operations/read_records.py b/temporalio/contrib/streams/system/operations/read_records.py index 398c1b4d6..d0cb4e460 100644 --- a/temporalio/contrib/streams/system/operations/read_records.py +++ b/temporalio/contrib/streams/system/operations/read_records.py @@ -35,6 +35,7 @@ async def read_records( start: temporalio.api.streamservice.v1.stream_service_pb2.ReadStart.ValueType | None = None, topics: list[str] | None = None, + topic_prefixes: list[str] | None = None, max_records: int | None = None, wait: datetime.timedelta | None = None, create_wait: datetime.timedelta | None = None, @@ -46,7 +47,9 @@ async def read_records( from_offset: Start at this offset. Empty uses start. INVALID_ARGUMENT if past the newest offset + 1. start: Where to start when from-offset is empty. Unspecified is the beginning. - topics: Empty reads every Topic. + topics: Topics to read, by exact match. With topic-prefixes, empty reads every + Topic. + topic_prefixes: Also read the Topics that start with one of these. max_records: At most this many Records. 1 to 1000. wait: How long to wait for a new Record before an empty answer. At most 60 s. create_wait: How long to wait for an uncreated Stream. Empty fails at once. At @@ -59,6 +62,7 @@ async def read_records( from_offset=from_offset, start=start, topics=topics or [], + topic_prefixes=topic_prefixes or [], max_records=max_records, wait=wait, create_wait=create_wait, diff --git a/tests/contrib/streams/fake_extension.py b/tests/contrib/streams/fake_extension.py index 38cfe6480..5c9dc5bde 100644 --- a/tests/contrib/streams/fake_extension.py +++ b/tests/contrib/streams/fake_extension.py @@ -124,9 +124,10 @@ async def _read( for record in stream.records if record.offset >= offset and ( - not request.topics + (not request.topics and not request.topic_prefixes) or record.kind != stream_pb.StreamRecordKind.STREAM_RECORD_KIND_DATA or record.topic in request.topics + or any(record.topic.startswith(p) for p in request.topic_prefixes) ) ] if found: diff --git a/tests/contrib/streams/test_facade.py b/tests/contrib/streams/test_facade.py index f30b8192a..d0e0c6416 100644 --- a/tests/contrib/streams/test_facade.py +++ b/tests/contrib/streams/test_facade.py @@ -321,3 +321,21 @@ async def test_read_default_type_for_untyped_topics(client: Client): assert isinstance(first, streams.Record) and first.data == Turn("model") # A typed Topic keeps its own type. assert isinstance(second, streams.Record) and isinstance(second.data, RawValue) + + +async def test_read_by_topic_prefix(client: Client): + client, _ = fake_client(client) + stream = await streams.create_stream(client, "prefixes") + for topic in ("agent/a", "agent/b", "other", "exact"): + stream.topic(topic).publish(topic) + await stream.flush() + read = [ + item.topic + async for item in stream.subscribe( + topics=[stream.topic("exact")], + topic_prefixes=["agent/"], + wait_new_records=False, + ) + if isinstance(item, streams.Record) + ] + assert read == ["agent/a", "agent/b", "exact"] diff --git a/tests/contrib/streams/test_redis.py b/tests/contrib/streams/test_redis.py index e7ef65e6d..813436c32 100644 --- a/tests/contrib/streams/test_redis.py +++ b/tests/contrib/streams/test_redis.py @@ -783,3 +783,20 @@ async def test_get_offset_of_a_created_empty_stream(stream_client: Client): # A Stream that nobody created is still an error. with pytest.raises(streams.StreamNotFoundError): await streams.for_handle(handle, name="missing").get_offset() + + +async def test_read_by_topic_prefix_on_redis(stream_client: Client): + stream = await streams.create_stream(stream_client, f"prefix-{uuid.uuid4()}") + for topic in ("agent/a", "agent/b", "other", "exact"): + stream.topic(topic, type=str).publish(topic) + await stream.flush() + read = [ + item.topic + async for item in stream.subscribe( + topics=[stream.topic("exact", type=str)], + topic_prefixes=["agent/"], + wait_new_records=False, + ) + if isinstance(item, streams.Record) + ] + assert read == ["agent/a", "agent/b", "exact"]