From 156e00315b756ba4a3656dbcd24f7c35f131af5c Mon Sep 17 00:00:00 2001 From: James Harris Date: Mon, 6 Jul 2026 07:57:10 +1000 Subject: [PATCH 01/11] Change `ConsumeAPI.ListStreams()` to use a streaming response. --- CHANGELOG.md | 6 +++ grpc/eventstreamgrpc/consume.pb.go | 49 +++++++++++------- grpc/eventstreamgrpc/consume.proto | 8 +-- grpc/eventstreamgrpc/consume_grpc.pb.go | 66 +++++++++++++----------- grpc/eventstreamgrpc/consume_primo.pb.go | 18 ++++--- 5 files changed, 87 insertions(+), 60 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b9daf2..90cd29f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,12 @@ The format is based on [Keep a Changelog], and this project adheres to [semantic versioning]: https://semver.org/spec/v2.0.0.html [bc]: https://github.com/dogmatiq/.github/blob/main/VERSIONING.md#changelogs +## [Unreleased] + +### Changed + +- **[BC]** Changed `ConsumeAPI.ListStreams()` to use a streaming response. + ## [0.26.5] - 2026-06-10 ### Changed diff --git a/grpc/eventstreamgrpc/consume.pb.go b/grpc/eventstreamgrpc/consume.pb.go index d71c26d..7e1b3e1 100644 --- a/grpc/eventstreamgrpc/consume.pb.go +++ b/grpc/eventstreamgrpc/consume.pb.go @@ -68,10 +68,10 @@ func (b0 ListStreamsRequest_builder) Build() *ListStreamsRequest { // ListStreamsResponse is the output of the ConsumeAPI.ListStreams method. type ListStreamsResponse struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_Streams *[]*Stream `protobuf:"bytes,1,rep,name=streams"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_Stream *Stream `protobuf:"bytes,1,opt,name=stream"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ListStreamsResponse) Reset() { @@ -99,31 +99,40 @@ func (x *ListStreamsResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *ListStreamsResponse) GetStreams() []*Stream { +func (x *ListStreamsResponse) GetStream() *Stream { if x != nil { - if x.xxx_hidden_Streams != nil { - return *x.xxx_hidden_Streams - } + return x.xxx_hidden_Stream } return nil } -func (x *ListStreamsResponse) SetStreams(v []*Stream) { - x.xxx_hidden_Streams = &v +func (x *ListStreamsResponse) SetStream(v *Stream) { + x.xxx_hidden_Stream = v +} + +func (x *ListStreamsResponse) HasStream() bool { + if x == nil { + return false + } + return x.xxx_hidden_Stream != nil +} + +func (x *ListStreamsResponse) ClearStream() { + x.xxx_hidden_Stream = nil } type ListStreamsResponse_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - // Streams is a list of event streams that can be consumed from this server. - Streams []*Stream + // Stream is an event stream that can be consumed from this server. + Stream *Stream } func (b0 ListStreamsResponse_builder) Build() *ListStreamsResponse { m0 := &ListStreamsResponse{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_Streams = &b.Streams + x.xxx_hidden_Stream = b.Stream return m0 } @@ -725,9 +734,9 @@ var File_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto protor const file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawDesc = "" + "\n" + "@github.com/dogmatiq/enginekit/grpc/eventstreamgrpc/consume.proto\x12\x1cdogma.eventstream.consume.v1\x1a@github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto\x1a8github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto\"\x14\n" + - "\x12ListStreamsRequest\"U\n" + - "\x13ListStreamsResponse\x12>\n" + - "\astreams\x18\x01 \x03(\v2$.dogma.eventstream.consume.v1.StreamR\astreams\"w\n" + + "\x12ListStreamsRequest\"S\n" + + "\x13ListStreamsResponse\x12<\n" + + "\x06stream\x18\x01 \x01(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"w\n" + "\x06Stream\x121\n" + "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x12:\n" + "\x0eevent_type_ids\x18\x02 \x03(\v2\x14.dogma.protobuf.UUIDR\feventTypeIds\"\xa4\x01\n" + @@ -745,10 +754,10 @@ const file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawD "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\"\x0e\n" + "\fNoEventTypes\"Q\n" + "\x15UnrecognizedEventType\x128\n" + - "\revent_type_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\veventTypeId2\xfc\x01\n" + + "\revent_type_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\veventTypeId2\xfe\x01\n" + "\n" + - "ConsumeAPI\x12r\n" + - "\vListStreams\x120.dogma.eventstream.consume.v1.ListStreamsRequest\x1a1.dogma.eventstream.consume.v1.ListStreamsResponse\x12z\n" + + "ConsumeAPI\x12t\n" + + "\vListStreams\x120.dogma.eventstream.consume.v1.ListStreamsRequest\x1a1.dogma.eventstream.consume.v1.ListStreamsResponse0\x01\x12z\n" + "\rConsumeEvents\x122.dogma.eventstream.consume.v1.ConsumeEventsRequest\x1a3.dogma.eventstream.consume.v1.ConsumeEventsResponse0\x01B4Z2github.com/dogmatiq/enginekit/grpc/eventstreamgrpcb\beditionsp\xe9\a" var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes = make([]protoimpl.MessageInfo, 9) @@ -766,7 +775,7 @@ var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_goType (*envelopepb.Envelope)(nil), // 10: dogma.protobuf.Envelope } var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_depIdxs = []int32{ - 2, // 0: dogma.eventstream.consume.v1.ListStreamsResponse.streams:type_name -> dogma.eventstream.consume.v1.Stream + 2, // 0: dogma.eventstream.consume.v1.ListStreamsResponse.stream:type_name -> dogma.eventstream.consume.v1.Stream 9, // 1: dogma.eventstream.consume.v1.Stream.stream_id:type_name -> dogma.protobuf.UUID 9, // 2: dogma.eventstream.consume.v1.Stream.event_type_ids:type_name -> dogma.protobuf.UUID 9, // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest.stream_id:type_name -> dogma.protobuf.UUID diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index 451c260..a44c817 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -11,7 +11,9 @@ import "github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto"; // stream. service ConsumeAPI { // ListStreams lists the streams that the server provides. - rpc ListStreams(ListStreamsRequest) returns (ListStreamsResponse); + // + // The server may add additional streams at any time. + rpc ListStreams(ListStreamsRequest) returns (stream ListStreamsResponse); // Consume starts consuming from a specific offset within an event stream. // @@ -39,8 +41,8 @@ message ListStreamsRequest {} // ListStreamsResponse is the output of the ConsumeAPI.ListStreams method. message ListStreamsResponse { - // Streams is a list of event streams that can be consumed from this server. - repeated Stream streams = 1; + // Stream is an event stream that can be consumed from this server. + Stream stream = 1; } // Stream describes an offset-based ordered event stream. diff --git a/grpc/eventstreamgrpc/consume_grpc.pb.go b/grpc/eventstreamgrpc/consume_grpc.pb.go index d53eb7e..cbd5790 100644 --- a/grpc/eventstreamgrpc/consume_grpc.pb.go +++ b/grpc/eventstreamgrpc/consume_grpc.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc v6.33.1 // source: github.com/dogmatiq/enginekit/grpc/eventstreamgrpc/consume.proto @@ -31,7 +31,9 @@ const ( // stream. type ConsumeAPIClient interface { // ListStreams lists the streams that the server provides. - ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (*ListStreamsResponse, error) + // + // The server may add additional streams at any time. + ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListStreamsResponse], error) // Consume starts consuming from a specific offset within an event stream. // // If the requested stream ID is unknown to the server it MUST return a @@ -60,19 +62,28 @@ func NewConsumeAPIClient(cc grpc.ClientConnInterface) ConsumeAPIClient { return &consumeAPIClient{cc} } -func (c *consumeAPIClient) ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (*ListStreamsResponse, error) { +func (c *consumeAPIClient) ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListStreamsResponse], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - out := new(ListStreamsResponse) - err := c.cc.Invoke(ctx, ConsumeAPI_ListStreams_FullMethodName, in, out, cOpts...) + stream, err := c.cc.NewStream(ctx, &ConsumeAPI_ServiceDesc.Streams[0], ConsumeAPI_ListStreams_FullMethodName, cOpts...) if err != nil { return nil, err } - return out, nil + x := &grpc.GenericClientStream[ListStreamsRequest, ListStreamsResponse]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil } +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type ConsumeAPI_ListStreamsClient = grpc.ServerStreamingClient[ListStreamsResponse] + func (c *consumeAPIClient) ConsumeEvents(ctx context.Context, in *ConsumeEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ConsumeEventsResponse], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &ConsumeAPI_ServiceDesc.Streams[0], ConsumeAPI_ConsumeEvents_FullMethodName, cOpts...) + stream, err := c.cc.NewStream(ctx, &ConsumeAPI_ServiceDesc.Streams[1], ConsumeAPI_ConsumeEvents_FullMethodName, cOpts...) if err != nil { return nil, err } @@ -97,7 +108,9 @@ type ConsumeAPI_ConsumeEventsClient = grpc.ServerStreamingClient[ConsumeEventsRe // stream. type ConsumeAPIServer interface { // ListStreams lists the streams that the server provides. - ListStreams(context.Context, *ListStreamsRequest) (*ListStreamsResponse, error) + // + // The server may add additional streams at any time. + ListStreams(*ListStreamsRequest, grpc.ServerStreamingServer[ListStreamsResponse]) error // Consume starts consuming from a specific offset within an event stream. // // If the requested stream ID is unknown to the server it MUST return a @@ -125,8 +138,8 @@ type ConsumeAPIServer interface { // pointer dereference when methods are called. type UnimplementedConsumeAPIServer struct{} -func (UnimplementedConsumeAPIServer) ListStreams(context.Context, *ListStreamsRequest) (*ListStreamsResponse, error) { - return nil, status.Error(codes.Unimplemented, "method ListStreams not implemented") +func (UnimplementedConsumeAPIServer) ListStreams(*ListStreamsRequest, grpc.ServerStreamingServer[ListStreamsResponse]) error { + return status.Error(codes.Unimplemented, "method ListStreams not implemented") } func (UnimplementedConsumeAPIServer) ConsumeEvents(*ConsumeEventsRequest, grpc.ServerStreamingServer[ConsumeEventsResponse]) error { return status.Error(codes.Unimplemented, "method ConsumeEvents not implemented") @@ -151,24 +164,17 @@ func RegisterConsumeAPIServer(s grpc.ServiceRegistrar, srv ConsumeAPIServer) { s.RegisterService(&ConsumeAPI_ServiceDesc, srv) } -func _ConsumeAPI_ListStreams_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(ListStreamsRequest) - if err := dec(in); err != nil { - return nil, err - } - if interceptor == nil { - return srv.(ConsumeAPIServer).ListStreams(ctx, in) - } - info := &grpc.UnaryServerInfo{ - Server: srv, - FullMethod: ConsumeAPI_ListStreams_FullMethodName, - } - handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(ConsumeAPIServer).ListStreams(ctx, req.(*ListStreamsRequest)) +func _ConsumeAPI_ListStreams_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(ListStreamsRequest) + if err := stream.RecvMsg(m); err != nil { + return err } - return interceptor(ctx, in, info, handler) + return srv.(ConsumeAPIServer).ListStreams(m, &grpc.GenericServerStream[ListStreamsRequest, ListStreamsResponse]{ServerStream: stream}) } +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type ConsumeAPI_ListStreamsServer = grpc.ServerStreamingServer[ListStreamsResponse] + func _ConsumeAPI_ConsumeEvents_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(ConsumeEventsRequest) if err := stream.RecvMsg(m); err != nil { @@ -186,13 +192,13 @@ type ConsumeAPI_ConsumeEventsServer = grpc.ServerStreamingServer[ConsumeEventsRe var ConsumeAPI_ServiceDesc = grpc.ServiceDesc{ ServiceName: "dogma.eventstream.consume.v1.ConsumeAPI", HandlerType: (*ConsumeAPIServer)(nil), - Methods: []grpc.MethodDesc{ + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ { - MethodName: "ListStreams", - Handler: _ConsumeAPI_ListStreams_Handler, + StreamName: "ListStreams", + Handler: _ConsumeAPI_ListStreams_Handler, + ServerStreams: true, }, - }, - Streams: []grpc.StreamDesc{ { StreamName: "ConsumeEvents", Handler: _ConsumeAPI_ConsumeEvents_Handler, diff --git a/grpc/eventstreamgrpc/consume_primo.pb.go b/grpc/eventstreamgrpc/consume_primo.pb.go index 1056076..abddf7c 100644 --- a/grpc/eventstreamgrpc/consume_primo.pb.go +++ b/grpc/eventstreamgrpc/consume_primo.pb.go @@ -69,7 +69,9 @@ func NewListStreamsResponseBuilder() *ListStreamsResponseBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *ListStreamsResponseBuilder) From(x *ListStreamsResponse) *ListStreamsResponseBuilder { proto.Reset(&b.prototype) - b.prototype.SetStreams(x.GetStreams()) + if x.HasStream() { + b.prototype.SetStream(x.GetStream()) + } return b } @@ -79,14 +81,16 @@ func (b *ListStreamsResponseBuilder) From(x *ListStreamsResponse) *ListStreamsRe // not modify previously constructed messages. func (b *ListStreamsResponseBuilder) Build() *ListStreamsResponse { m := &ListStreamsResponse{} - m.SetStreams(b.prototype.GetStreams()) + if b.prototype.HasStream() { + m.SetStream(b.prototype.GetStream()) + } return m } -// WithStreams configures the builder to set the Streams field to v, +// WithStream configures the builder to set the Stream field to v, // then returns b. -func (b *ListStreamsResponseBuilder) WithStreams(v []*Stream) *ListStreamsResponseBuilder { - b.prototype.SetStreams(v) +func (b *ListStreamsResponseBuilder) WithStream(v *Stream) *ListStreamsResponseBuilder { + b.prototype.SetStream(v) return b } @@ -473,14 +477,14 @@ var _ ConsumeAPIClient = (*ConsumeAPIClientStub)(nil) type ConsumeAPIClientStub struct { // ListStreamsFunc is a function that implements the ListStreams RPC method. - ListStreamsFunc func(ctx context.Context, req *ListStreamsRequest, options ...grpc.CallOption) (*ListStreamsResponse, error) + ListStreamsFunc func(ctx context.Context, req *ListStreamsRequest, options ...grpc.CallOption) (ConsumeAPI_ListStreamsClient, error) // ConsumeEventsFunc is a function that implements the ConsumeEvents RPC method. ConsumeEventsFunc func(ctx context.Context, req *ConsumeEventsRequest, options ...grpc.CallOption) (ConsumeAPI_ConsumeEventsClient, error) } // ListStreams calls c.ListStreamsFunc if it is non-nil. Otherwise, it returns an error. -func (c *ConsumeAPIClientStub) ListStreams(ctx context.Context, req *ListStreamsRequest, options ...grpc.CallOption) (*ListStreamsResponse, error) { +func (c *ConsumeAPIClientStub) ListStreams(ctx context.Context, req *ListStreamsRequest, options ...grpc.CallOption) (ConsumeAPI_ListStreamsClient, error) { if c.ListStreamsFunc == nil { return nil, status.Error(codes.Unimplemented, "method ListStreams not implemented") } From aa996b1c87487dd676f88c486544c040da92f86b Mon Sep 17 00:00:00 2001 From: James Harris Date: Mon, 6 Jul 2026 08:04:20 +1000 Subject: [PATCH 02/11] Fix doc comment. --- grpc/eventstreamgrpc/consume.proto | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index a44c817..a153421 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -15,7 +15,8 @@ service ConsumeAPI { // The server may add additional streams at any time. rpc ListStreams(ListStreamsRequest) returns (stream ListStreamsResponse); - // Consume starts consuming from a specific offset within an event stream. + // ConsumeEvents starts consuming from a specific offset within an event + // stream. // // If the requested stream ID is unknown to the server it MUST return a // NOT_FOUND error with an attached [UnrecognizedStream] value. See From 90b46bcff2958620f12b7b2736889751b4ec4a8b Mon Sep 17 00:00:00 2001 From: James Harris Date: Wed, 8 Jul 2026 13:48:28 +1000 Subject: [PATCH 03/11] Refactor `dogma.eventstream.consume.v1.ConsumeAPI` to support new stream/event discovery. --- grpc/eventstreamgrpc/consume.go | 57 -- grpc/eventstreamgrpc/consume.pb.go | 632 ++++++----------------- grpc/eventstreamgrpc/consume.proto | 136 ++--- grpc/eventstreamgrpc/consume_grpc.pb.go | 66 +-- grpc/eventstreamgrpc/consume_primo.pb.go | 365 ++----------- 5 files changed, 301 insertions(+), 955 deletions(-) delete mode 100644 grpc/eventstreamgrpc/consume.go diff --git a/grpc/eventstreamgrpc/consume.go b/grpc/eventstreamgrpc/consume.go deleted file mode 100644 index 5eeb6bb..0000000 --- a/grpc/eventstreamgrpc/consume.go +++ /dev/null @@ -1,57 +0,0 @@ -package eventstreamgrpc - -import ( - uuidpb "github.com/dogmatiq/enginekit/protobuf/uuidpb" - codes "google.golang.org/grpc/codes" - status "google.golang.org/grpc/status" -) - -// UnrecognizedStreamError returns an error indicating that the given stream is -// not recognized by the server. -func UnrecognizedStreamError(id *uuidpb.UUID) error { - s, err := status. - New(codes.NotFound, "unrecognized stream"). - WithDetails( - NewUnrecognizedStreamBuilder(). - WithStreamId(id). - Build(), - ) - - if err != nil { - panic(err) - } - - return s.Err() -} - -// NoEventTypesError returns an error indicating that the that the client made a -// consume request without specifying any event types. -func NoEventTypesError() error { - s, err := status. - New(codes.InvalidArgument, "no event types specified"). - WithDetails(&NoEventTypes{}) - - if err != nil { - panic(err) - } - - return s.Err() -} - -// UnrecognizedEventTypeError returns an error indicating that the given event -// type is not recognized by the server. -func UnrecognizedEventTypeError(id *uuidpb.UUID) error { - s, err := status. - New(codes.InvalidArgument, "unrecognized event type"). - WithDetails( - NewUnrecognizedEventTypeBuilder(). - WithEventTypeId(id). - Build(), - ) - - if err != nil { - panic(err) - } - - return s.Err() -} diff --git a/grpc/eventstreamgrpc/consume.pb.go b/grpc/eventstreamgrpc/consume.pb.go index 7e1b3e1..b5dc657 100644 --- a/grpc/eventstreamgrpc/consume.pb.go +++ b/grpc/eventstreamgrpc/consume.pb.go @@ -22,11 +22,12 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// ListStreamsRequest is the input to the ConsumeAPI.ListStreams method. +// ListStreamsRequest is the input to the [ConsumeAPI.ListStreams] method. type ListStreamsRequest struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_MessageTypeIds *[]*uuidpb.UUID `protobuf:"bytes,1,rep,name=message_type_ids,json=messageTypeIds"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ListStreamsRequest) Reset() { @@ -54,22 +55,39 @@ func (x *ListStreamsRequest) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } +func (x *ListStreamsRequest) GetMessageTypeIds() []*uuidpb.UUID { + if x != nil { + if x.xxx_hidden_MessageTypeIds != nil { + return *x.xxx_hidden_MessageTypeIds + } + } + return nil +} + +func (x *ListStreamsRequest) SetMessageTypeIds(v []*uuidpb.UUID) { + x.xxx_hidden_MessageTypeIds = &v +} + type ListStreamsRequest_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. + // MessageTypeIds is the set of message type IDs of the events that the client + // is interested in consuming. + MessageTypeIds []*uuidpb.UUID } func (b0 ListStreamsRequest_builder) Build() *ListStreamsRequest { m0 := &ListStreamsRequest{} b, x := &b0, m0 _, _ = b, x + x.xxx_hidden_MessageTypeIds = &b.MessageTypeIds return m0 } -// ListStreamsResponse is the output of the ConsumeAPI.ListStreams method. +// ListStreamsResponse is the output of the [ConsumeAPI.ListStreams] method. type ListStreamsResponse struct { state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_Stream *Stream `protobuf:"bytes,1,opt,name=stream"` + xxx_hidden_Stream *[]*Stream `protobuf:"bytes,1,rep,name=stream"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -99,50 +117,41 @@ func (x *ListStreamsResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *ListStreamsResponse) GetStream() *Stream { +func (x *ListStreamsResponse) GetStream() []*Stream { if x != nil { - return x.xxx_hidden_Stream + if x.xxx_hidden_Stream != nil { + return *x.xxx_hidden_Stream + } } return nil } -func (x *ListStreamsResponse) SetStream(v *Stream) { - x.xxx_hidden_Stream = v -} - -func (x *ListStreamsResponse) HasStream() bool { - if x == nil { - return false - } - return x.xxx_hidden_Stream != nil -} - -func (x *ListStreamsResponse) ClearStream() { - x.xxx_hidden_Stream = nil +func (x *ListStreamsResponse) SetStream(v []*Stream) { + x.xxx_hidden_Stream = &v } type ListStreamsResponse_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - // Stream is an event stream that can be consumed from this server. - Stream *Stream + // Streams is a list of event streams that can be consumed from this server. + Stream []*Stream } func (b0 ListStreamsResponse_builder) Build() *ListStreamsResponse { m0 := &ListStreamsResponse{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_Stream = b.Stream + x.xxx_hidden_Stream = &b.Stream return m0 } // Stream describes an offset-based ordered event stream. type Stream struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_StreamId *uuidpb.UUID `protobuf:"bytes,1,opt,name=stream_id,json=streamId"` - xxx_hidden_EventTypeIds *[]*uuidpb.UUID `protobuf:"bytes,2,rep,name=event_type_ids,json=eventTypeIds"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_StreamId *uuidpb.UUID `protobuf:"bytes,1,opt,name=stream_id,json=streamId"` + xxx_hidden_NextOffset uint64 `protobuf:"varint,2,opt,name=next_offset,json=nextOffset"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Stream) Reset() { @@ -177,21 +186,19 @@ func (x *Stream) GetStreamId() *uuidpb.UUID { return nil } -func (x *Stream) GetEventTypeIds() []*uuidpb.UUID { +func (x *Stream) GetNextOffset() uint64 { if x != nil { - if x.xxx_hidden_EventTypeIds != nil { - return *x.xxx_hidden_EventTypeIds - } + return x.xxx_hidden_NextOffset } - return nil + return 0 } func (x *Stream) SetStreamId(v *uuidpb.UUID) { x.xxx_hidden_StreamId = v } -func (x *Stream) SetEventTypeIds(v []*uuidpb.UUID) { - x.xxx_hidden_EventTypeIds = &v +func (x *Stream) SetNextOffset(v uint64) { + x.xxx_hidden_NextOffset = v } func (x *Stream) HasStreamId() bool { @@ -210,8 +217,11 @@ type Stream_builder struct { // StreamId is a unique identifier for the stream. StreamId *uuidpb.UUID - // EventTypeIds is the set of type IDs of events that may appear on the stream. - EventTypeIds []*uuidpb.UUID + // NextOffset is the offset of the next event that will be written to the + // stream. The first event in the stream is at offset zero; therefore, this + // value is equivalent to the number of events that have been written to the + // stream. + NextOffset uint64 } func (b0 Stream_builder) Build() *Stream { @@ -219,18 +229,18 @@ func (b0 Stream_builder) Build() *Stream { b, x := &b0, m0 _, _ = b, x x.xxx_hidden_StreamId = b.StreamId - x.xxx_hidden_EventTypeIds = &b.EventTypeIds + x.xxx_hidden_NextOffset = b.NextOffset return m0 } -// ConsumeEventsRequest is the input to the ConsumeAPI.ConsumeEvents method. +// ConsumeEventsRequest is the input to the [ConsumeAPI.ConsumeEvents] method. type ConsumeEventsRequest struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_StreamId *uuidpb.UUID `protobuf:"bytes,1,opt,name=stream_id,json=streamId"` - xxx_hidden_Offset uint64 `protobuf:"varint,2,opt,name=offset"` - xxx_hidden_EventTypeIds *[]*uuidpb.UUID `protobuf:"bytes,3,rep,name=event_type_ids,json=eventTypeIds"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_StreamId *uuidpb.UUID `protobuf:"bytes,1,opt,name=stream_id,json=streamId"` + xxx_hidden_CheckpointOffset uint64 `protobuf:"varint,2,opt,name=checkpoint_offset,json=checkpointOffset"` + xxx_hidden_MessageTypeIds *[]*uuidpb.UUID `protobuf:"bytes,3,rep,name=message_type_ids,json=messageTypeIds"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ConsumeEventsRequest) Reset() { @@ -265,17 +275,17 @@ func (x *ConsumeEventsRequest) GetStreamId() *uuidpb.UUID { return nil } -func (x *ConsumeEventsRequest) GetOffset() uint64 { +func (x *ConsumeEventsRequest) GetCheckpointOffset() uint64 { if x != nil { - return x.xxx_hidden_Offset + return x.xxx_hidden_CheckpointOffset } return 0 } -func (x *ConsumeEventsRequest) GetEventTypeIds() []*uuidpb.UUID { +func (x *ConsumeEventsRequest) GetMessageTypeIds() []*uuidpb.UUID { if x != nil { - if x.xxx_hidden_EventTypeIds != nil { - return *x.xxx_hidden_EventTypeIds + if x.xxx_hidden_MessageTypeIds != nil { + return *x.xxx_hidden_MessageTypeIds } } return nil @@ -285,12 +295,12 @@ func (x *ConsumeEventsRequest) SetStreamId(v *uuidpb.UUID) { x.xxx_hidden_StreamId = v } -func (x *ConsumeEventsRequest) SetOffset(v uint64) { - x.xxx_hidden_Offset = v +func (x *ConsumeEventsRequest) SetCheckpointOffset(v uint64) { + x.xxx_hidden_CheckpointOffset = v } -func (x *ConsumeEventsRequest) SetEventTypeIds(v []*uuidpb.UUID) { - x.xxx_hidden_EventTypeIds = &v +func (x *ConsumeEventsRequest) SetMessageTypeIds(v []*uuidpb.UUID) { + x.xxx_hidden_MessageTypeIds = &v } func (x *ConsumeEventsRequest) HasStreamId() bool { @@ -309,12 +319,14 @@ type ConsumeEventsRequest_builder struct { // StreamId is the ID from which events are consumed. StreamId *uuidpb.UUID - // Offset is the offset of the earliest event to be consumed. - Offset uint64 - // EventTypeIds is a list of type IDs of the events to be consumed. The - // consumer must be explicit about the event types that it understands; there - // is no mechanism to request all event types. - EventTypeIds []*uuidpb.UUID + // CheckpointOffset is the offset at which the consumer begins. + CheckpointOffset uint64 + // MessageTypeIds is the set of message type IDs of the events that the client + // is interested in consuming. + // + // The server MUST omit any events that do not match one of the specified + // message type IDs from responses. + MessageTypeIds []*uuidpb.UUID } func (b0 ConsumeEventsRequest_builder) Build() *ConsumeEventsRequest { @@ -322,18 +334,30 @@ func (b0 ConsumeEventsRequest_builder) Build() *ConsumeEventsRequest { b, x := &b0, m0 _, _ = b, x x.xxx_hidden_StreamId = b.StreamId - x.xxx_hidden_Offset = b.Offset - x.xxx_hidden_EventTypeIds = &b.EventTypeIds + x.xxx_hidden_CheckpointOffset = b.CheckpointOffset + x.xxx_hidden_MessageTypeIds = &b.MessageTypeIds return m0 } -// ConsumeResponse is the (streaming) output of the ConsumeAPI.ConsumeEvents -// method. +// ConsumeEventsResponse is the (streaming) output of the +// [ConsumeAPI.ConsumeEvents] method. +// +// It contains all of the events that match the requested message type IDs +// specified in the [ConsumeEventsRequest], starting at the prior checkpoint +// offset up to, but not including, the checkpoint offset in this response. +// +// If this is the first response in the consume operation, the prior checkpoint +// offset is the checkpoint offset in the [ConsumeEventsRequest]. Otherwise, it +// is the checkpoint offset in the prior response. +// +// The list of events may be empty, indicating that there are no matching events +// in that range. type ConsumeEventsResponse struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_Operation isConsumeEventsResponse_Operation `protobuf_oneof:"operation"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_CheckpointOffset uint64 `protobuf:"varint,1,opt,name=checkpoint_offset,json=checkpointOffset"` + xxx_hidden_Envelopes *[]*envelopepb.MultiEnvelope `protobuf:"bytes,2,rep,name=envelopes"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ConsumeEventsResponse) Reset() { @@ -361,371 +385,61 @@ func (x *ConsumeEventsResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *ConsumeEventsResponse) GetEventDelivery() *ConsumeEventsResponse_EventDelivery { +func (x *ConsumeEventsResponse) GetCheckpointOffset() uint64 { if x != nil { - if x, ok := x.xxx_hidden_Operation.(*consumeEventsResponse_EventDelivery_); ok { - return x.EventDelivery - } - } - return nil -} - -func (x *ConsumeEventsResponse) SetEventDelivery(v *ConsumeEventsResponse_EventDelivery) { - if v == nil { - x.xxx_hidden_Operation = nil - return - } - x.xxx_hidden_Operation = &consumeEventsResponse_EventDelivery_{v} -} - -func (x *ConsumeEventsResponse) HasOperation() bool { - if x == nil { - return false - } - return x.xxx_hidden_Operation != nil -} - -func (x *ConsumeEventsResponse) HasEventDelivery() bool { - if x == nil { - return false - } - _, ok := x.xxx_hidden_Operation.(*consumeEventsResponse_EventDelivery_) - return ok -} - -func (x *ConsumeEventsResponse) ClearOperation() { - x.xxx_hidden_Operation = nil -} - -func (x *ConsumeEventsResponse) ClearEventDelivery() { - if _, ok := x.xxx_hidden_Operation.(*consumeEventsResponse_EventDelivery_); ok { - x.xxx_hidden_Operation = nil - } -} - -const ConsumeEventsResponse_Operation_not_set_case case_ConsumeEventsResponse_Operation = 0 -const ConsumeEventsResponse_EventDelivery_case case_ConsumeEventsResponse_Operation = 1 - -func (x *ConsumeEventsResponse) WhichOperation() case_ConsumeEventsResponse_Operation { - if x == nil { - return ConsumeEventsResponse_Operation_not_set_case - } - switch x.xxx_hidden_Operation.(type) { - case *consumeEventsResponse_EventDelivery_: - return ConsumeEventsResponse_EventDelivery_case - default: - return ConsumeEventsResponse_Operation_not_set_case - } -} - -type ConsumeEventsResponse_builder struct { - _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - - // Fields of oneof xxx_hidden_Operation: - EventDelivery *ConsumeEventsResponse_EventDelivery - // -- end of xxx_hidden_Operation -} - -func (b0 ConsumeEventsResponse_builder) Build() *ConsumeEventsResponse { - m0 := &ConsumeEventsResponse{} - b, x := &b0, m0 - _, _ = b, x - if b.EventDelivery != nil { - x.xxx_hidden_Operation = &consumeEventsResponse_EventDelivery_{b.EventDelivery} - } - return m0 -} - -type case_ConsumeEventsResponse_Operation protoreflect.FieldNumber - -func (x case_ConsumeEventsResponse_Operation) String() string { - md := file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[4].Descriptor() - if x == 0 { - return "not set" - } - return protoimpl.X.MessageFieldStringOf(md, protoreflect.FieldNumber(x)) -} - -type isConsumeEventsResponse_Operation interface { - isConsumeEventsResponse_Operation() -} - -type consumeEventsResponse_EventDelivery_ struct { - EventDelivery *ConsumeEventsResponse_EventDelivery `protobuf:"bytes,1,opt,name=event_delivery,json=eventDelivery,oneof"` -} - -func (*consumeEventsResponse_EventDelivery_) isConsumeEventsResponse_Operation() {} - -// UnrecognizedStream is an error-details value for INVALID_ARGUMENT errors that -// occurred because a consumer requested an unrecognized stream ID. -type UnrecognizedStream struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_StreamId *uuidpb.UUID `protobuf:"bytes,1,opt,name=stream_id,json=streamId"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *UnrecognizedStream) Reset() { - *x = UnrecognizedStream{} - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[5] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *UnrecognizedStream) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*UnrecognizedStream) ProtoMessage() {} - -func (x *UnrecognizedStream) ProtoReflect() protoreflect.Message { - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[5] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -func (x *UnrecognizedStream) GetStreamId() *uuidpb.UUID { - if x != nil { - return x.xxx_hidden_StreamId - } - return nil -} - -func (x *UnrecognizedStream) SetStreamId(v *uuidpb.UUID) { - x.xxx_hidden_StreamId = v -} - -func (x *UnrecognizedStream) HasStreamId() bool { - if x == nil { - return false - } - return x.xxx_hidden_StreamId != nil -} - -func (x *UnrecognizedStream) ClearStreamId() { - x.xxx_hidden_StreamId = nil -} - -type UnrecognizedStream_builder struct { - _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - - // ApplicationKey is the ID of the unrecognized stream. - StreamId *uuidpb.UUID -} - -func (b0 UnrecognizedStream_builder) Build() *UnrecognizedStream { - m0 := &UnrecognizedStream{} - b, x := &b0, m0 - _, _ = b, x - x.xxx_hidden_StreamId = b.StreamId - return m0 -} - -// NoEventTypes is an error-details value for INVALID_ARGUMENT errors that -// occurred because a client sent a consume request without specifying any event -// type IDs. -type NoEventTypes struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *NoEventTypes) Reset() { - *x = NoEventTypes{} - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[6] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *NoEventTypes) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*NoEventTypes) ProtoMessage() {} - -func (x *NoEventTypes) ProtoReflect() protoreflect.Message { - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[6] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -type NoEventTypes_builder struct { - _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - -} - -func (b0 NoEventTypes_builder) Build() *NoEventTypes { - m0 := &NoEventTypes{} - b, x := &b0, m0 - _, _ = b, x - return m0 -} - -// UnrecognizedEventType is an error-details value for INVALID_ARGUMENT errors -// that occurred because a specific event type was not recognized by the server. -type UnrecognizedEventType struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_EventTypeId *uuidpb.UUID `protobuf:"bytes,1,opt,name=event_type_id,json=eventTypeId"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *UnrecognizedEventType) Reset() { - *x = UnrecognizedEventType{} - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[7] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *UnrecognizedEventType) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*UnrecognizedEventType) ProtoMessage() {} - -func (x *UnrecognizedEventType) ProtoReflect() protoreflect.Message { - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[7] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -func (x *UnrecognizedEventType) GetEventTypeId() *uuidpb.UUID { - if x != nil { - return x.xxx_hidden_EventTypeId - } - return nil -} - -func (x *UnrecognizedEventType) SetEventTypeId(v *uuidpb.UUID) { - x.xxx_hidden_EventTypeId = v -} - -func (x *UnrecognizedEventType) HasEventTypeId() bool { - if x == nil { - return false - } - return x.xxx_hidden_EventTypeId != nil -} - -func (x *UnrecognizedEventType) ClearEventTypeId() { - x.xxx_hidden_EventTypeId = nil -} - -type UnrecognizedEventType_builder struct { - _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - - // EventTypeId is the ID of the unrecognized event type. - EventTypeId *uuidpb.UUID -} - -func (b0 UnrecognizedEventType_builder) Build() *UnrecognizedEventType { - m0 := &UnrecognizedEventType{} - b, x := &b0, m0 - _, _ = b, x - x.xxx_hidden_EventTypeId = b.EventTypeId - return m0 -} - -// EventDelivery represents the delivery of a single event to the consumer. -type ConsumeEventsResponse_EventDelivery struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_Offset uint64 `protobuf:"varint,1,opt,name=offset"` - xxx_hidden_Envelope *envelopepb.Envelope `protobuf:"bytes,2,opt,name=envelope"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *ConsumeEventsResponse_EventDelivery) Reset() { - *x = ConsumeEventsResponse_EventDelivery{} - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[8] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *ConsumeEventsResponse_EventDelivery) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*ConsumeEventsResponse_EventDelivery) ProtoMessage() {} - -func (x *ConsumeEventsResponse_EventDelivery) ProtoReflect() protoreflect.Message { - mi := &file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[8] - if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -func (x *ConsumeEventsResponse_EventDelivery) GetOffset() uint64 { - if x != nil { - return x.xxx_hidden_Offset + return x.xxx_hidden_CheckpointOffset } return 0 } -func (x *ConsumeEventsResponse_EventDelivery) GetEnvelope() *envelopepb.Envelope { +func (x *ConsumeEventsResponse) GetEnvelopes() []*envelopepb.MultiEnvelope { if x != nil { - return x.xxx_hidden_Envelope + if x.xxx_hidden_Envelopes != nil { + return *x.xxx_hidden_Envelopes + } } return nil } -func (x *ConsumeEventsResponse_EventDelivery) SetOffset(v uint64) { - x.xxx_hidden_Offset = v -} - -func (x *ConsumeEventsResponse_EventDelivery) SetEnvelope(v *envelopepb.Envelope) { - x.xxx_hidden_Envelope = v -} - -func (x *ConsumeEventsResponse_EventDelivery) HasEnvelope() bool { - if x == nil { - return false - } - return x.xxx_hidden_Envelope != nil +func (x *ConsumeEventsResponse) SetCheckpointOffset(v uint64) { + x.xxx_hidden_CheckpointOffset = v } -func (x *ConsumeEventsResponse_EventDelivery) ClearEnvelope() { - x.xxx_hidden_Envelope = nil +func (x *ConsumeEventsResponse) SetEnvelopes(v []*envelopepb.MultiEnvelope) { + x.xxx_hidden_Envelopes = &v } -type ConsumeEventsResponse_EventDelivery_builder struct { +type ConsumeEventsResponse_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - // Offset is the event's offset within the stream. - Offset uint64 - // Envelope is the envelope containing the event. - Envelope *envelopepb.Envelope + // CheckpointOffset is the offset immediately after the range of the stream + // described by this response. + // + // After handling this and all preceding responses, the client may use this + // value as the checkpoint offset to resume consuming from the stream in a + // subsequent [ConsumeEventsRequest]. + CheckpointOffset uint64 + // Envelopes is a (possibly empty) list of multi-envelopes containing events + // at offsets between the prior checkpoint offset and the checkpoint offset + // in this response. + // + // Each envelope includes the [envelopepb.EventStreamPosition] extension. + // + // The envelopes are ordered by the offset of the events they contain. There + // is no additional relationship implied between envelopes that appear in + // the same response. + // + // The offsets of the events within this response, or even within a single + // [envelopepb.MultiEnvelope] are not necessarily contiguous. + Envelopes []*envelopepb.MultiEnvelope } -func (b0 ConsumeEventsResponse_EventDelivery_builder) Build() *ConsumeEventsResponse_EventDelivery { - m0 := &ConsumeEventsResponse_EventDelivery{} +func (b0 ConsumeEventsResponse_builder) Build() *ConsumeEventsResponse { + m0 := &ConsumeEventsResponse{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_Offset = b.Offset - x.xxx_hidden_Envelope = b.Envelope + x.xxx_hidden_CheckpointOffset = b.CheckpointOffset + x.xxx_hidden_Envelopes = &b.Envelopes return m0 } @@ -733,66 +447,53 @@ var File_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto protor const file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawDesc = "" + "\n" + - "@github.com/dogmatiq/enginekit/grpc/eventstreamgrpc/consume.proto\x12\x1cdogma.eventstream.consume.v1\x1a@github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto\x1a8github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto\"\x14\n" + - "\x12ListStreamsRequest\"S\n" + + "@github.com/dogmatiq/enginekit/grpc/eventstreamgrpc/consume.proto\x12\x1cdogma.eventstream.consume.v1\x1a@github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto\x1a8github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto\"T\n" + + "\x12ListStreamsRequest\x12>\n" + + "\x10message_type_ids\x18\x01 \x03(\v2\x14.dogma.protobuf.UUIDR\x0emessageTypeIds\"S\n" + "\x13ListStreamsResponse\x12<\n" + - "\x06stream\x18\x01 \x01(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"w\n" + + "\x06stream\x18\x01 \x03(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"c\n" + "\x06Stream\x121\n" + - "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x12:\n" + - "\x0eevent_type_ids\x18\x02 \x03(\v2\x14.dogma.protobuf.UUIDR\feventTypeIds\"\xa4\x01\n" + + "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x12&\n" + + "\vnext_offset\x18\x02 \x01(\x04B\x05\xaa\x01\x02\b\x02R\n" + + "nextOffset\"\xbd\x01\n" + "\x14ConsumeEventsRequest\x121\n" + - "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x12\x1d\n" + - "\x06offset\x18\x02 \x01(\x04B\x05\xaa\x01\x02\b\x02R\x06offset\x12:\n" + - "\x0eevent_type_ids\x18\x03 \x03(\v2\x14.dogma.protobuf.UUIDR\feventTypeIds\"\xf6\x01\n" + - "\x15ConsumeEventsResponse\x12j\n" + - "\x0eevent_delivery\x18\x01 \x01(\v2A.dogma.eventstream.consume.v1.ConsumeEventsResponse.EventDeliveryH\x00R\reventDelivery\x1ad\n" + - "\rEventDelivery\x12\x1d\n" + - "\x06offset\x18\x01 \x01(\x04B\x05\xaa\x01\x02\b\x02R\x06offset\x124\n" + - "\benvelope\x18\x02 \x01(\v2\x18.dogma.protobuf.EnvelopeR\benvelopeB\v\n" + - "\toperation\"G\n" + - "\x12UnrecognizedStream\x121\n" + - "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\"\x0e\n" + - "\fNoEventTypes\"Q\n" + - "\x15UnrecognizedEventType\x128\n" + - "\revent_type_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\veventTypeId2\xfe\x01\n" + + "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x122\n" + + "\x11checkpoint_offset\x18\x02 \x01(\x04B\x05\xaa\x01\x02\b\x02R\x10checkpointOffset\x12>\n" + + "\x10message_type_ids\x18\x03 \x03(\v2\x14.dogma.protobuf.UUIDR\x0emessageTypeIds\"\x88\x01\n" + + "\x15ConsumeEventsResponse\x122\n" + + "\x11checkpoint_offset\x18\x01 \x01(\x04B\x05\xaa\x01\x02\b\x02R\x10checkpointOffset\x12;\n" + + "\tenvelopes\x18\x02 \x03(\v2\x1d.dogma.protobuf.MultiEnvelopeR\tenvelopes2\xfe\x01\n" + "\n" + "ConsumeAPI\x12t\n" + "\vListStreams\x120.dogma.eventstream.consume.v1.ListStreamsRequest\x1a1.dogma.eventstream.consume.v1.ListStreamsResponse0\x01\x12z\n" + "\rConsumeEvents\x122.dogma.eventstream.consume.v1.ConsumeEventsRequest\x1a3.dogma.eventstream.consume.v1.ConsumeEventsResponse0\x01B4Z2github.com/dogmatiq/enginekit/grpc/eventstreamgrpcb\beditionsp\xe9\a" -var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes = make([]protoimpl.MessageInfo, 9) +var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes = make([]protoimpl.MessageInfo, 5) var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_goTypes = []any{ - (*ListStreamsRequest)(nil), // 0: dogma.eventstream.consume.v1.ListStreamsRequest - (*ListStreamsResponse)(nil), // 1: dogma.eventstream.consume.v1.ListStreamsResponse - (*Stream)(nil), // 2: dogma.eventstream.consume.v1.Stream - (*ConsumeEventsRequest)(nil), // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest - (*ConsumeEventsResponse)(nil), // 4: dogma.eventstream.consume.v1.ConsumeEventsResponse - (*UnrecognizedStream)(nil), // 5: dogma.eventstream.consume.v1.UnrecognizedStream - (*NoEventTypes)(nil), // 6: dogma.eventstream.consume.v1.NoEventTypes - (*UnrecognizedEventType)(nil), // 7: dogma.eventstream.consume.v1.UnrecognizedEventType - (*ConsumeEventsResponse_EventDelivery)(nil), // 8: dogma.eventstream.consume.v1.ConsumeEventsResponse.EventDelivery - (*uuidpb.UUID)(nil), // 9: dogma.protobuf.UUID - (*envelopepb.Envelope)(nil), // 10: dogma.protobuf.Envelope + (*ListStreamsRequest)(nil), // 0: dogma.eventstream.consume.v1.ListStreamsRequest + (*ListStreamsResponse)(nil), // 1: dogma.eventstream.consume.v1.ListStreamsResponse + (*Stream)(nil), // 2: dogma.eventstream.consume.v1.Stream + (*ConsumeEventsRequest)(nil), // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest + (*ConsumeEventsResponse)(nil), // 4: dogma.eventstream.consume.v1.ConsumeEventsResponse + (*uuidpb.UUID)(nil), // 5: dogma.protobuf.UUID + (*envelopepb.MultiEnvelope)(nil), // 6: dogma.protobuf.MultiEnvelope } var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_depIdxs = []int32{ - 2, // 0: dogma.eventstream.consume.v1.ListStreamsResponse.stream:type_name -> dogma.eventstream.consume.v1.Stream - 9, // 1: dogma.eventstream.consume.v1.Stream.stream_id:type_name -> dogma.protobuf.UUID - 9, // 2: dogma.eventstream.consume.v1.Stream.event_type_ids:type_name -> dogma.protobuf.UUID - 9, // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest.stream_id:type_name -> dogma.protobuf.UUID - 9, // 4: dogma.eventstream.consume.v1.ConsumeEventsRequest.event_type_ids:type_name -> dogma.protobuf.UUID - 8, // 5: dogma.eventstream.consume.v1.ConsumeEventsResponse.event_delivery:type_name -> dogma.eventstream.consume.v1.ConsumeEventsResponse.EventDelivery - 9, // 6: dogma.eventstream.consume.v1.UnrecognizedStream.stream_id:type_name -> dogma.protobuf.UUID - 9, // 7: dogma.eventstream.consume.v1.UnrecognizedEventType.event_type_id:type_name -> dogma.protobuf.UUID - 10, // 8: dogma.eventstream.consume.v1.ConsumeEventsResponse.EventDelivery.envelope:type_name -> dogma.protobuf.Envelope - 0, // 9: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:input_type -> dogma.eventstream.consume.v1.ListStreamsRequest - 3, // 10: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:input_type -> dogma.eventstream.consume.v1.ConsumeEventsRequest - 1, // 11: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:output_type -> dogma.eventstream.consume.v1.ListStreamsResponse - 4, // 12: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:output_type -> dogma.eventstream.consume.v1.ConsumeEventsResponse - 11, // [11:13] is the sub-list for method output_type - 9, // [9:11] is the sub-list for method input_type - 9, // [9:9] is the sub-list for extension type_name - 9, // [9:9] is the sub-list for extension extendee - 0, // [0:9] is the sub-list for field type_name + 5, // 0: dogma.eventstream.consume.v1.ListStreamsRequest.message_type_ids:type_name -> dogma.protobuf.UUID + 2, // 1: dogma.eventstream.consume.v1.ListStreamsResponse.stream:type_name -> dogma.eventstream.consume.v1.Stream + 5, // 2: dogma.eventstream.consume.v1.Stream.stream_id:type_name -> dogma.protobuf.UUID + 5, // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest.stream_id:type_name -> dogma.protobuf.UUID + 5, // 4: dogma.eventstream.consume.v1.ConsumeEventsRequest.message_type_ids:type_name -> dogma.protobuf.UUID + 6, // 5: dogma.eventstream.consume.v1.ConsumeEventsResponse.envelopes:type_name -> dogma.protobuf.MultiEnvelope + 0, // 6: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:input_type -> dogma.eventstream.consume.v1.ListStreamsRequest + 3, // 7: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:input_type -> dogma.eventstream.consume.v1.ConsumeEventsRequest + 1, // 8: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:output_type -> dogma.eventstream.consume.v1.ListStreamsResponse + 4, // 9: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:output_type -> dogma.eventstream.consume.v1.ConsumeEventsResponse + 8, // [8:10] is the sub-list for method output_type + 6, // [6:8] is the sub-list for method input_type + 6, // [6:6] is the sub-list for extension type_name + 6, // [6:6] is the sub-list for extension extendee + 0, // [0:6] is the sub-list for field type_name } func init() { file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_init() } @@ -800,16 +501,13 @@ func file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_init( if File_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto != nil { return } - file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_msgTypes[4].OneofWrappers = []any{ - (*consumeEventsResponse_EventDelivery_)(nil), - } type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawDesc), len(file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawDesc)), NumEnums: 0, - NumMessages: 9, + NumMessages: 5, NumExtensions: 0, NumServices: 1, }, diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index a153421..2a9eaa7 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -7,43 +7,41 @@ import "github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto"; import "github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto"; -// ConsumeAPI is a service for consuming events from an offset-based ordered -// stream. +// ConsumeAPI is a service for consuming event messages from a offset-based, +// ordered event streams. service ConsumeAPI { - // ListStreams lists the streams that the server provides. - // - // The server may add additional streams at any time. + // ListStreams returns a the event streams that may contain events whose + // message type IDs are among those specified in the [ListStreamsRequest]. rpc ListStreams(ListStreamsRequest) returns (stream ListStreamsResponse); - // ConsumeEvents starts consuming from a specific offset within an event - // stream. + // ConsumeEvents returns events in order, starting from a specific offset + // within an event stream. // - // If the requested stream ID is unknown to the server it MUST return a - // NOT_FOUND error with an attached [UnrecognizedStream] value. See - // [UnrecognizedStreamError]. + // The server delivers every event at or after the requested checkpoint offset + // whose message type ID is among those specified in the + // [ConsumeEventsRequest]. If no such events remain in the event stream, and + // the server cannot produce new events with any of the requested message type + // IDs, it closes the gRPC stream. // - // If the requested offset is beyond the end of the stream, the server SHOULD - // keep the stream open and send new events as they are written to the stream. + // If the requested stream ID is not recognized a NOT_FOUND error occurs. // - // The requested type IDs MUST be a subset of those type IDs associated with - // the stream, as per the result of the ListStreams operation. If any other - // type IDs are requested the server MUST return an INVALID_ARGUMENT error - // with an attached [UnrecognizedEventType] value for each unrecognized type - // ID. See [UnrecognizedEventTypeError]. - // - // If no type IDs are specified the server MUST return an INVALID_ARGUMENT - // error with an attached [NoEventTypes] value. See [NoEventTypesError]. + // If the requested offset is beyond the end of the stream an OUT_OF_RANGE + // error occurs. rpc ConsumeEvents(ConsumeEventsRequest) returns (stream ConsumeEventsResponse); } -// ListStreamsRequest is the input to the ConsumeAPI.ListStreams method. -message ListStreamsRequest {} +// ListStreamsRequest is the input to the [ConsumeAPI.ListStreams] method. +message ListStreamsRequest { + // MessageTypeIds is the set of message type IDs of the events that the client + // is interested in consuming. + repeated dogma.protobuf.UUID message_type_ids = 1; +} -// ListStreamsResponse is the output of the ConsumeAPI.ListStreams method. +// ListStreamsResponse is the output of the [ConsumeAPI.ListStreams] method. message ListStreamsResponse { - // Stream is an event stream that can be consumed from this server. - Stream stream = 1; + // Streams is a list of event streams that can be consumed from this server. + repeated Stream stream = 1; } // Stream describes an offset-based ordered event stream. @@ -51,54 +49,62 @@ message Stream { // StreamId is a unique identifier for the stream. dogma.protobuf.UUID stream_id = 1; - // EventTypeIds is the set of type IDs of events that may appear on the stream. - repeated dogma.protobuf.UUID event_type_ids = 2; + // NextOffset is the offset of the next event that will be written to the + // stream. The first event in the stream is at offset zero; therefore, this + // value is equivalent to the number of events that have been written to the + // stream. + uint64 next_offset = 2 [features.field_presence = IMPLICIT]; } -// ConsumeEventsRequest is the input to the ConsumeAPI.ConsumeEvents method. +// ConsumeEventsRequest is the input to the [ConsumeAPI.ConsumeEvents] method. message ConsumeEventsRequest { // StreamId is the ID from which events are consumed. dogma.protobuf.UUID stream_id = 1; - // Offset is the offset of the earliest event to be consumed. - uint64 offset = 2 [features.field_presence = IMPLICIT]; + // CheckpointOffset is the offset at which the consumer begins. + uint64 checkpoint_offset = 2 [features.field_presence = IMPLICIT]; - // EventTypeIds is a list of type IDs of the events to be consumed. The - // consumer must be explicit about the event types that it understands; there - // is no mechanism to request all event types. - repeated dogma.protobuf.UUID event_type_ids = 3; + // MessageTypeIds is the set of message type IDs of the events that the client + // is interested in consuming. + // + // The server MUST omit any events that do not match one of the specified + // message type IDs from responses. + repeated dogma.protobuf.UUID message_type_ids = 3; } -// ConsumeResponse is the (streaming) output of the ConsumeAPI.ConsumeEvents -// method. +// ConsumeEventsResponse is the (streaming) output of the +// [ConsumeAPI.ConsumeEvents] method. +// +// It contains all of the events that match the requested message type IDs +// specified in the [ConsumeEventsRequest], starting at the prior checkpoint +// offset up to, but not including, the checkpoint offset in this response. +// +// If this is the first response in the consume operation, the prior checkpoint +// offset is the checkpoint offset in the [ConsumeEventsRequest]. Otherwise, it +// is the checkpoint offset in the prior response. +// +// The list of events may be empty, indicating that there are no matching events +// in that range. message ConsumeEventsResponse { - // EventDelivery represents the delivery of a single event to the consumer. - message EventDelivery { - // Offset is the event's offset within the stream. - uint64 offset = 1 [features.field_presence = IMPLICIT]; - - // Envelope is the envelope containing the event. - dogma.protobuf.Envelope envelope = 2; - } - - oneof operation { EventDelivery event_delivery = 1; } -} - -// UnrecognizedStream is an error-details value for INVALID_ARGUMENT errors that -// occurred because a consumer requested an unrecognized stream ID. -message UnrecognizedStream { - // ApplicationKey is the ID of the unrecognized stream. - dogma.protobuf.UUID stream_id = 1; -} - -// NoEventTypes is an error-details value for INVALID_ARGUMENT errors that -// occurred because a client sent a consume request without specifying any event -// type IDs. -message NoEventTypes {} - -// UnrecognizedEventType is an error-details value for INVALID_ARGUMENT errors -// that occurred because a specific event type was not recognized by the server. -message UnrecognizedEventType { - // EventTypeId is the ID of the unrecognized event type. - dogma.protobuf.UUID event_type_id = 1; + // CheckpointOffset is the offset immediately after the range of the stream + // described by this response. + // + // After handling this and all preceding responses, the client may use this + // value as the checkpoint offset to resume consuming from the stream in a + // subsequent [ConsumeEventsRequest]. + uint64 checkpoint_offset = 1 [features.field_presence = IMPLICIT]; + + // Envelopes is a (possibly empty) list of multi-envelopes containing events + // at offsets between the prior checkpoint offset and the checkpoint offset + // in this response. + // + // Each envelope includes the [envelopepb.EventStreamPosition] extension. + // + // The envelopes are ordered by the offset of the events they contain. There + // is no additional relationship implied between envelopes that appear in + // the same response. + // + // The offsets of the events within this response, or even within a single + // [envelopepb.MultiEnvelope] are not necessarily contiguous. + repeated dogma.protobuf.MultiEnvelope envelopes = 2; } diff --git a/grpc/eventstreamgrpc/consume_grpc.pb.go b/grpc/eventstreamgrpc/consume_grpc.pb.go index cbd5790..e946aaa 100644 --- a/grpc/eventstreamgrpc/consume_grpc.pb.go +++ b/grpc/eventstreamgrpc/consume_grpc.pb.go @@ -27,30 +27,25 @@ const ( // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. // -// ConsumeAPI is a service for consuming events from an offset-based ordered -// stream. +// ConsumeAPI is a service for consuming event messages from a offset-based, +// ordered event streams. type ConsumeAPIClient interface { - // ListStreams lists the streams that the server provides. - // - // The server may add additional streams at any time. + // ListStreams returns a the event streams that may contain events whose + // message type IDs are among those specified in the [ListStreamsRequest]. ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListStreamsResponse], error) - // Consume starts consuming from a specific offset within an event stream. - // - // If the requested stream ID is unknown to the server it MUST return a - // NOT_FOUND error with an attached [UnrecognizedStream] value. See - // [UnrecognizedStreamError]. + // ConsumeEvents returns events in order, starting from a specific offset + // within an event stream. // - // If the requested offset is beyond the end of the stream, the server SHOULD - // keep the stream open and send new events as they are written to the stream. + // The server delivers every event at or after the requested checkpoint offset + // whose message type ID is among those specified in the + // [ConsumeEventsRequest]. If no such events remain in the event stream, and + // the server cannot produce new events with any of the requested message type + // IDs, it closes the gRPC stream. // - // The requested type IDs MUST be a subset of those type IDs associated with - // the stream, as per the result of the ListStreams operation. If any other - // type IDs are requested the server MUST return an INVALID_ARGUMENT error - // with an attached [UnrecognizedEventType] value for each unrecognized type - // ID. See [UnrecognizedEventTypeError]. + // If the requested stream ID is not recognized a NOT_FOUND error occurs. // - // If no type IDs are specified the server MUST return an INVALID_ARGUMENT - // error with an attached [NoEventTypes] value. See [NoEventTypesError]. + // If the requested offset is beyond the end of the stream an OUT_OF_RANGE + // error occurs. ConsumeEvents(ctx context.Context, in *ConsumeEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ConsumeEventsResponse], error) } @@ -104,30 +99,25 @@ type ConsumeAPI_ConsumeEventsClient = grpc.ServerStreamingClient[ConsumeEventsRe // All implementations should embed UnimplementedConsumeAPIServer // for forward compatibility. // -// ConsumeAPI is a service for consuming events from an offset-based ordered -// stream. +// ConsumeAPI is a service for consuming event messages from a offset-based, +// ordered event streams. type ConsumeAPIServer interface { - // ListStreams lists the streams that the server provides. - // - // The server may add additional streams at any time. + // ListStreams returns a the event streams that may contain events whose + // message type IDs are among those specified in the [ListStreamsRequest]. ListStreams(*ListStreamsRequest, grpc.ServerStreamingServer[ListStreamsResponse]) error - // Consume starts consuming from a specific offset within an event stream. - // - // If the requested stream ID is unknown to the server it MUST return a - // NOT_FOUND error with an attached [UnrecognizedStream] value. See - // [UnrecognizedStreamError]. + // ConsumeEvents returns events in order, starting from a specific offset + // within an event stream. // - // If the requested offset is beyond the end of the stream, the server SHOULD - // keep the stream open and send new events as they are written to the stream. + // The server delivers every event at or after the requested checkpoint offset + // whose message type ID is among those specified in the + // [ConsumeEventsRequest]. If no such events remain in the event stream, and + // the server cannot produce new events with any of the requested message type + // IDs, it closes the gRPC stream. // - // The requested type IDs MUST be a subset of those type IDs associated with - // the stream, as per the result of the ListStreams operation. If any other - // type IDs are requested the server MUST return an INVALID_ARGUMENT error - // with an attached [UnrecognizedEventType] value for each unrecognized type - // ID. See [UnrecognizedEventTypeError]. + // If the requested stream ID is not recognized a NOT_FOUND error occurs. // - // If no type IDs are specified the server MUST return an INVALID_ARGUMENT - // error with an attached [NoEventTypes] value. See [NoEventTypesError]. + // If the requested offset is beyond the end of the stream an OUT_OF_RANGE + // error occurs. ConsumeEvents(*ConsumeEventsRequest, grpc.ServerStreamingServer[ConsumeEventsResponse]) error } diff --git a/grpc/eventstreamgrpc/consume_primo.pb.go b/grpc/eventstreamgrpc/consume_primo.pb.go index abddf7c..11f4da7 100644 --- a/grpc/eventstreamgrpc/consume_primo.pb.go +++ b/grpc/eventstreamgrpc/consume_primo.pb.go @@ -16,15 +16,6 @@ import ( proto "google.golang.org/protobuf/proto" ) -// TryGetEventDelivery returns the EventDelivery value if x.Operation is set to EventDelivery. -// Otherwise, ok is false and v is the zero-value. -func (x *ConsumeEventsResponse) TryGetEventDelivery() (v *ConsumeEventsResponse_EventDelivery, ok bool) { - if x.HasEventDelivery() { - return x.GetEventDelivery(), true - } - return v, false -} - type ListStreamsRequestBuilder struct { prototype ListStreamsRequest } @@ -41,6 +32,7 @@ func NewListStreamsRequestBuilder() *ListStreamsRequestBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *ListStreamsRequestBuilder) From(x *ListStreamsRequest) *ListStreamsRequestBuilder { proto.Reset(&b.prototype) + b.prototype.SetMessageTypeIds(x.GetMessageTypeIds()) return b } @@ -50,9 +42,17 @@ func (b *ListStreamsRequestBuilder) From(x *ListStreamsRequest) *ListStreamsRequ // not modify previously constructed messages. func (b *ListStreamsRequestBuilder) Build() *ListStreamsRequest { m := &ListStreamsRequest{} + m.SetMessageTypeIds(b.prototype.GetMessageTypeIds()) return m } +// WithMessageTypeIds configures the builder to set the MessageTypeIds field to v, +// then returns b. +func (b *ListStreamsRequestBuilder) WithMessageTypeIds(v []*uuidpb.UUID) *ListStreamsRequestBuilder { + b.prototype.SetMessageTypeIds(v) + return b +} + type ListStreamsResponseBuilder struct { prototype ListStreamsResponse } @@ -69,9 +69,7 @@ func NewListStreamsResponseBuilder() *ListStreamsResponseBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *ListStreamsResponseBuilder) From(x *ListStreamsResponse) *ListStreamsResponseBuilder { proto.Reset(&b.prototype) - if x.HasStream() { - b.prototype.SetStream(x.GetStream()) - } + b.prototype.SetStream(x.GetStream()) return b } @@ -81,15 +79,13 @@ func (b *ListStreamsResponseBuilder) From(x *ListStreamsResponse) *ListStreamsRe // not modify previously constructed messages. func (b *ListStreamsResponseBuilder) Build() *ListStreamsResponse { m := &ListStreamsResponse{} - if b.prototype.HasStream() { - m.SetStream(b.prototype.GetStream()) - } + m.SetStream(b.prototype.GetStream()) return m } // WithStream configures the builder to set the Stream field to v, // then returns b. -func (b *ListStreamsResponseBuilder) WithStream(v *Stream) *ListStreamsResponseBuilder { +func (b *ListStreamsResponseBuilder) WithStream(v []*Stream) *ListStreamsResponseBuilder { b.prototype.SetStream(v) return b } @@ -113,7 +109,7 @@ func (b *StreamBuilder) From(x *Stream) *StreamBuilder { if x.HasStreamId() { b.prototype.SetStreamId(x.GetStreamId()) } - b.prototype.SetEventTypeIds(x.GetEventTypeIds()) + b.prototype.SetNextOffset(x.GetNextOffset()) return b } @@ -126,7 +122,7 @@ func (b *StreamBuilder) Build() *Stream { if b.prototype.HasStreamId() { m.SetStreamId(b.prototype.GetStreamId()) } - m.SetEventTypeIds(b.prototype.GetEventTypeIds()) + m.SetNextOffset(b.prototype.GetNextOffset()) return m } @@ -137,10 +133,10 @@ func (b *StreamBuilder) WithStreamId(v *uuidpb.UUID) *StreamBuilder { return b } -// WithEventTypeIds configures the builder to set the EventTypeIds field to v, +// WithNextOffset configures the builder to set the NextOffset field to v, // then returns b. -func (b *StreamBuilder) WithEventTypeIds(v []*uuidpb.UUID) *StreamBuilder { - b.prototype.SetEventTypeIds(v) +func (b *StreamBuilder) WithNextOffset(v uint64) *StreamBuilder { + b.prototype.SetNextOffset(v) return b } @@ -163,8 +159,8 @@ func (b *ConsumeEventsRequestBuilder) From(x *ConsumeEventsRequest) *ConsumeEven if x.HasStreamId() { b.prototype.SetStreamId(x.GetStreamId()) } - b.prototype.SetOffset(x.GetOffset()) - b.prototype.SetEventTypeIds(x.GetEventTypeIds()) + b.prototype.SetCheckpointOffset(x.GetCheckpointOffset()) + b.prototype.SetMessageTypeIds(x.GetMessageTypeIds()) return b } @@ -177,8 +173,8 @@ func (b *ConsumeEventsRequestBuilder) Build() *ConsumeEventsRequest { if b.prototype.HasStreamId() { m.SetStreamId(b.prototype.GetStreamId()) } - m.SetOffset(b.prototype.GetOffset()) - m.SetEventTypeIds(b.prototype.GetEventTypeIds()) + m.SetCheckpointOffset(b.prototype.GetCheckpointOffset()) + m.SetMessageTypeIds(b.prototype.GetMessageTypeIds()) return m } @@ -189,17 +185,17 @@ func (b *ConsumeEventsRequestBuilder) WithStreamId(v *uuidpb.UUID) *ConsumeEvent return b } -// WithOffset configures the builder to set the Offset field to v, +// WithCheckpointOffset configures the builder to set the CheckpointOffset field to v, // then returns b. -func (b *ConsumeEventsRequestBuilder) WithOffset(v uint64) *ConsumeEventsRequestBuilder { - b.prototype.SetOffset(v) +func (b *ConsumeEventsRequestBuilder) WithCheckpointOffset(v uint64) *ConsumeEventsRequestBuilder { + b.prototype.SetCheckpointOffset(v) return b } -// WithEventTypeIds configures the builder to set the EventTypeIds field to v, +// WithMessageTypeIds configures the builder to set the MessageTypeIds field to v, // then returns b. -func (b *ConsumeEventsRequestBuilder) WithEventTypeIds(v []*uuidpb.UUID) *ConsumeEventsRequestBuilder { - b.prototype.SetEventTypeIds(v) +func (b *ConsumeEventsRequestBuilder) WithMessageTypeIds(v []*uuidpb.UUID) *ConsumeEventsRequestBuilder { + b.prototype.SetMessageTypeIds(v) return b } @@ -219,9 +215,8 @@ func NewConsumeEventsResponseBuilder() *ConsumeEventsResponseBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *ConsumeEventsResponseBuilder) From(x *ConsumeEventsResponse) *ConsumeEventsResponseBuilder { proto.Reset(&b.prototype) - if x.HasEventDelivery() { - b.prototype.SetEventDelivery(x.GetEventDelivery()) - } + b.prototype.SetCheckpointOffset(x.GetCheckpointOffset()) + b.prototype.SetEnvelopes(x.GetEnvelopes()) return b } @@ -231,247 +226,25 @@ func (b *ConsumeEventsResponseBuilder) From(x *ConsumeEventsResponse) *ConsumeEv // not modify previously constructed messages. func (b *ConsumeEventsResponseBuilder) Build() *ConsumeEventsResponse { m := &ConsumeEventsResponse{} - if b.prototype.HasEventDelivery() { - m.SetEventDelivery(b.prototype.GetEventDelivery()) - } - return m -} - -// WithEventDelivery configures the builder to set the EventDelivery field to v, -// then returns b. -func (b *ConsumeEventsResponseBuilder) WithEventDelivery(v *ConsumeEventsResponse_EventDelivery) *ConsumeEventsResponseBuilder { - b.prototype.SetEventDelivery(v) - return b -} - -type ConsumeEventsResponse_EventDeliveryBuilder struct { - prototype ConsumeEventsResponse_EventDelivery -} - -// NewConsumeEventsResponse_EventDeliveryBuilder returns a builder that constructs [ConsumeEventsResponse_EventDelivery] messages. -func NewConsumeEventsResponse_EventDeliveryBuilder() *ConsumeEventsResponse_EventDeliveryBuilder { - return &ConsumeEventsResponse_EventDeliveryBuilder{} -} - -// From configures the builder to use x as the prototype for new messages, -// then returns b. -// -// It performs a shallow copy of x, such that any changes made via the builder -// do not modify x. It does not make a copy of the field values themselves. -func (b *ConsumeEventsResponse_EventDeliveryBuilder) From(x *ConsumeEventsResponse_EventDelivery) *ConsumeEventsResponse_EventDeliveryBuilder { - proto.Reset(&b.prototype) - b.prototype.SetOffset(x.GetOffset()) - if x.HasEnvelope() { - b.prototype.SetEnvelope(x.GetEnvelope()) - } - return b -} - -// Build returns a new [ConsumeEventsResponse_EventDelivery] containing the values configured via the builder. -// -// Each call returns a new message, such that future changes to the builder do -// not modify previously constructed messages. -func (b *ConsumeEventsResponse_EventDeliveryBuilder) Build() *ConsumeEventsResponse_EventDelivery { - m := &ConsumeEventsResponse_EventDelivery{} - m.SetOffset(b.prototype.GetOffset()) - if b.prototype.HasEnvelope() { - m.SetEnvelope(b.prototype.GetEnvelope()) - } - return m -} - -// WithOffset configures the builder to set the Offset field to v, -// then returns b. -func (b *ConsumeEventsResponse_EventDeliveryBuilder) WithOffset(v uint64) *ConsumeEventsResponse_EventDeliveryBuilder { - b.prototype.SetOffset(v) - return b -} - -// WithEnvelope configures the builder to set the Envelope field to v, -// then returns b. -func (b *ConsumeEventsResponse_EventDeliveryBuilder) WithEnvelope(v *envelopepb.Envelope) *ConsumeEventsResponse_EventDeliveryBuilder { - b.prototype.SetEnvelope(v) - return b -} - -type UnrecognizedStreamBuilder struct { - prototype UnrecognizedStream -} - -// NewUnrecognizedStreamBuilder returns a builder that constructs [UnrecognizedStream] messages. -func NewUnrecognizedStreamBuilder() *UnrecognizedStreamBuilder { - return &UnrecognizedStreamBuilder{} -} - -// From configures the builder to use x as the prototype for new messages, -// then returns b. -// -// It performs a shallow copy of x, such that any changes made via the builder -// do not modify x. It does not make a copy of the field values themselves. -func (b *UnrecognizedStreamBuilder) From(x *UnrecognizedStream) *UnrecognizedStreamBuilder { - proto.Reset(&b.prototype) - if x.HasStreamId() { - b.prototype.SetStreamId(x.GetStreamId()) - } - return b -} - -// Build returns a new [UnrecognizedStream] containing the values configured via the builder. -// -// Each call returns a new message, such that future changes to the builder do -// not modify previously constructed messages. -func (b *UnrecognizedStreamBuilder) Build() *UnrecognizedStream { - m := &UnrecognizedStream{} - if b.prototype.HasStreamId() { - m.SetStreamId(b.prototype.GetStreamId()) - } + m.SetCheckpointOffset(b.prototype.GetCheckpointOffset()) + m.SetEnvelopes(b.prototype.GetEnvelopes()) return m } -// WithStreamId configures the builder to set the StreamId field to v, +// WithCheckpointOffset configures the builder to set the CheckpointOffset field to v, // then returns b. -func (b *UnrecognizedStreamBuilder) WithStreamId(v *uuidpb.UUID) *UnrecognizedStreamBuilder { - b.prototype.SetStreamId(v) +func (b *ConsumeEventsResponseBuilder) WithCheckpointOffset(v uint64) *ConsumeEventsResponseBuilder { + b.prototype.SetCheckpointOffset(v) return b } -type NoEventTypesBuilder struct { - prototype NoEventTypes -} - -// NewNoEventTypesBuilder returns a builder that constructs [NoEventTypes] messages. -func NewNoEventTypesBuilder() *NoEventTypesBuilder { - return &NoEventTypesBuilder{} -} - -// From configures the builder to use x as the prototype for new messages, +// WithEnvelopes configures the builder to set the Envelopes field to v, // then returns b. -// -// It performs a shallow copy of x, such that any changes made via the builder -// do not modify x. It does not make a copy of the field values themselves. -func (b *NoEventTypesBuilder) From(x *NoEventTypes) *NoEventTypesBuilder { - proto.Reset(&b.prototype) +func (b *ConsumeEventsResponseBuilder) WithEnvelopes(v []*envelopepb.MultiEnvelope) *ConsumeEventsResponseBuilder { + b.prototype.SetEnvelopes(v) return b } -// Build returns a new [NoEventTypes] containing the values configured via the builder. -// -// Each call returns a new message, such that future changes to the builder do -// not modify previously constructed messages. -func (b *NoEventTypesBuilder) Build() *NoEventTypes { - m := &NoEventTypes{} - return m -} - -type UnrecognizedEventTypeBuilder struct { - prototype UnrecognizedEventType -} - -// NewUnrecognizedEventTypeBuilder returns a builder that constructs [UnrecognizedEventType] messages. -func NewUnrecognizedEventTypeBuilder() *UnrecognizedEventTypeBuilder { - return &UnrecognizedEventTypeBuilder{} -} - -// From configures the builder to use x as the prototype for new messages, -// then returns b. -// -// It performs a shallow copy of x, such that any changes made via the builder -// do not modify x. It does not make a copy of the field values themselves. -func (b *UnrecognizedEventTypeBuilder) From(x *UnrecognizedEventType) *UnrecognizedEventTypeBuilder { - proto.Reset(&b.prototype) - if x.HasEventTypeId() { - b.prototype.SetEventTypeId(x.GetEventTypeId()) - } - return b -} - -// Build returns a new [UnrecognizedEventType] containing the values configured via the builder. -// -// Each call returns a new message, such that future changes to the builder do -// not modify previously constructed messages. -func (b *UnrecognizedEventTypeBuilder) Build() *UnrecognizedEventType { - m := &UnrecognizedEventType{} - if b.prototype.HasEventTypeId() { - m.SetEventTypeId(b.prototype.GetEventTypeId()) - } - return m -} - -// WithEventTypeId configures the builder to set the EventTypeId field to v, -// then returns b. -func (b *UnrecognizedEventTypeBuilder) WithEventTypeId(v *uuidpb.UUID) *UnrecognizedEventTypeBuilder { - b.prototype.SetEventTypeId(v) - return b -} - -// MustSwitch_ConsumeEventsResponse_Operation invokes one of the given functions based on -// the value of x.Operation. -// -// It panics if x.Operation is nil. -func MustSwitch_ConsumeEventsResponse_Operation( - x *ConsumeEventsResponse, - caseEventDelivery func(*ConsumeEventsResponse_EventDelivery), -) { - switch x.WhichOperation() { - case ConsumeEventsResponse_EventDelivery_case: - caseEventDelivery(x.GetEventDelivery()) - default: - panic("MustSwitch_ConsumeEventsResponse_Operation: x.Operation is not set") - } -} - -// Switch_ConsumeEventsResponse_Operation invokes one of the given functions based on -// the value of x.Operation. -// -// It calls none() if x.Operation is nil. -func Switch_ConsumeEventsResponse_Operation( - x *ConsumeEventsResponse, - caseEventDelivery func(*ConsumeEventsResponse_EventDelivery), - none func(), -) { - switch x.WhichOperation() { - case ConsumeEventsResponse_EventDelivery_case: - caseEventDelivery(x.GetEventDelivery()) - default: - none() - } -} - -// MustMap_ConsumeEventsResponse_Operation maps x.Operation to a value of type T by invoking -// one of the given functions. -// -// It invokes the function that corresponds to the value of x.Operation, -// and returns that function's result. It panics if x.Operation is nil. -func MustMap_ConsumeEventsResponse_Operation[T any]( - x *ConsumeEventsResponse, - caseEventDelivery func(*ConsumeEventsResponse_EventDelivery) T, -) T { - switch x.WhichOperation() { - case ConsumeEventsResponse_EventDelivery_case: - return caseEventDelivery(x.GetEventDelivery()) - default: - panic("MustMap_ConsumeEventsResponse_Operation: x.Operation is not set") - } -} - -// Map_ConsumeEventsResponse_Operation maps x.Operation to a value of type T by invoking -// one of the given functions. -// -// It invokes the function that corresponds to the value of x.Operation, -// and returns that function's result. It calls none() if x.Operation is nil. -func Map_ConsumeEventsResponse_Operation[T any]( - x *ConsumeEventsResponse, - caseEventDelivery func(*ConsumeEventsResponse_EventDelivery) T, - none func() T, -) T { - switch x.WhichOperation() { - case ConsumeEventsResponse_EventDelivery_case: - return caseEventDelivery(x.GetEventDelivery()) - default: - return none() - } -} - // ConsumeAPIClientStub is a test implementation of the [ConsumeAPIClient] interface. var _ ConsumeAPIClient = (*ConsumeAPIClientStub)(nil) @@ -578,67 +351,3 @@ func (x *ConsumeEventsResponse) MarshalBinary() ([]byte, error) { func (x *ConsumeEventsResponse) UnmarshalBinary(data []byte) error { return proto.Unmarshal(data, x) } - -// MarshalBinary returns the binary representation of the message, equivalent to -// calling proto.Marshal(x). -// -// It allows [*ConsumeEventsResponse_EventDelivery] to implement [encoding.BinaryMarshaler]. -func (x *ConsumeEventsResponse_EventDelivery) MarshalBinary() ([]byte, error) { - return proto.Marshal(x) -} - -// UnmarshalBinary populates x from its binary representation, equivalent to -// calling proto.Unmarshal(data, x). -// -// It allows [*ConsumeEventsResponse_EventDelivery] to implement [encoding.BinaryUnmarshaler]. -func (x *ConsumeEventsResponse_EventDelivery) UnmarshalBinary(data []byte) error { - return proto.Unmarshal(data, x) -} - -// MarshalBinary returns the binary representation of the message, equivalent to -// calling proto.Marshal(x). -// -// It allows [*UnrecognizedStream] to implement [encoding.BinaryMarshaler]. -func (x *UnrecognizedStream) MarshalBinary() ([]byte, error) { - return proto.Marshal(x) -} - -// UnmarshalBinary populates x from its binary representation, equivalent to -// calling proto.Unmarshal(data, x). -// -// It allows [*UnrecognizedStream] to implement [encoding.BinaryUnmarshaler]. -func (x *UnrecognizedStream) UnmarshalBinary(data []byte) error { - return proto.Unmarshal(data, x) -} - -// MarshalBinary returns the binary representation of the message, equivalent to -// calling proto.Marshal(x). -// -// It allows [*NoEventTypes] to implement [encoding.BinaryMarshaler]. -func (x *NoEventTypes) MarshalBinary() ([]byte, error) { - return proto.Marshal(x) -} - -// UnmarshalBinary populates x from its binary representation, equivalent to -// calling proto.Unmarshal(data, x). -// -// It allows [*NoEventTypes] to implement [encoding.BinaryUnmarshaler]. -func (x *NoEventTypes) UnmarshalBinary(data []byte) error { - return proto.Unmarshal(data, x) -} - -// MarshalBinary returns the binary representation of the message, equivalent to -// calling proto.Marshal(x). -// -// It allows [*UnrecognizedEventType] to implement [encoding.BinaryMarshaler]. -func (x *UnrecognizedEventType) MarshalBinary() ([]byte, error) { - return proto.Marshal(x) -} - -// UnmarshalBinary populates x from its binary representation, equivalent to -// calling proto.Unmarshal(data, x). -// -// It allows [*UnrecognizedEventType] to implement [encoding.BinaryUnmarshaler]. -func (x *UnrecognizedEventType) UnmarshalBinary(data []byte) error { - return proto.Unmarshal(data, x) -} From 965f3b6316dc65ab49b3154f11f3f5898e847b64 Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 08:18:32 +1000 Subject: [PATCH 04/11] Update grpc/eventstreamgrpc/consume.proto Co-authored-by: Danil Petrov --- grpc/eventstreamgrpc/consume.proto | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index 2a9eaa7..c57ed0a 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -10,7 +10,7 @@ import "github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto"; // ConsumeAPI is a service for consuming event messages from a offset-based, // ordered event streams. service ConsumeAPI { - // ListStreams returns a the event streams that may contain events whose + // ListStreams returns the event streams that may contain events whose // message type IDs are among those specified in the [ListStreamsRequest]. rpc ListStreams(ListStreamsRequest) returns (stream ListStreamsResponse); From aa4b0920f7df8bc30a30aa7f5b3a9a1e5426cd6b Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 08:46:47 +1000 Subject: [PATCH 05/11] Fix grammar. --- grpc/eventstreamgrpc/consume.proto | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index c57ed0a..58cdc66 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -7,7 +7,7 @@ import "github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto"; import "github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto"; -// ConsumeAPI is a service for consuming event messages from a offset-based, +// ConsumeAPI is a service for consuming event messages from offset-based, // ordered event streams. service ConsumeAPI { // ListStreams returns the event streams that may contain events whose From 4a19962fa0bb882845a010a62ff1cacd099e6f06 Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 08:47:13 +1000 Subject: [PATCH 06/11] Regenerate proto files. --- grpc/eventstreamgrpc/consume_grpc.pb.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/grpc/eventstreamgrpc/consume_grpc.pb.go b/grpc/eventstreamgrpc/consume_grpc.pb.go index e946aaa..eaad355 100644 --- a/grpc/eventstreamgrpc/consume_grpc.pb.go +++ b/grpc/eventstreamgrpc/consume_grpc.pb.go @@ -27,10 +27,10 @@ const ( // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. // -// ConsumeAPI is a service for consuming event messages from a offset-based, +// ConsumeAPI is a service for consuming event messages from offset-based, // ordered event streams. type ConsumeAPIClient interface { - // ListStreams returns a the event streams that may contain events whose + // ListStreams returns the event streams that may contain events whose // message type IDs are among those specified in the [ListStreamsRequest]. ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListStreamsResponse], error) // ConsumeEvents returns events in order, starting from a specific offset @@ -99,10 +99,10 @@ type ConsumeAPI_ConsumeEventsClient = grpc.ServerStreamingClient[ConsumeEventsRe // All implementations should embed UnimplementedConsumeAPIServer // for forward compatibility. // -// ConsumeAPI is a service for consuming event messages from a offset-based, +// ConsumeAPI is a service for consuming event messages from offset-based, // ordered event streams. type ConsumeAPIServer interface { - // ListStreams returns a the event streams that may contain events whose + // ListStreams returns the event streams that may contain events whose // message type IDs are among those specified in the [ListStreamsRequest]. ListStreams(*ListStreamsRequest, grpc.ServerStreamingServer[ListStreamsResponse]) error // ConsumeEvents returns events in order, starting from a specific offset From bb162331ac41c9444997390b72e19e77c622743d Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 08:59:03 +1000 Subject: [PATCH 07/11] Change `ListStreamsResponse` to include a single stream for simplicity. --- grpc/eventstreamgrpc/consume.pb.go | 39 ++++++++++++++---------- grpc/eventstreamgrpc/consume.proto | 20 ++++++------ grpc/eventstreamgrpc/consume_grpc.pb.go | 16 +++++----- grpc/eventstreamgrpc/consume_primo.pb.go | 10 ++++-- 4 files changed, 47 insertions(+), 38 deletions(-) diff --git a/grpc/eventstreamgrpc/consume.pb.go b/grpc/eventstreamgrpc/consume.pb.go index b5dc657..d2a9675 100644 --- a/grpc/eventstreamgrpc/consume.pb.go +++ b/grpc/eventstreamgrpc/consume.pb.go @@ -87,7 +87,7 @@ func (b0 ListStreamsRequest_builder) Build() *ListStreamsRequest { // ListStreamsResponse is the output of the [ConsumeAPI.ListStreams] method. type ListStreamsResponse struct { state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_Stream *[]*Stream `protobuf:"bytes,1,rep,name=stream"` + xxx_hidden_Stream *Stream `protobuf:"bytes,1,opt,name=stream"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -117,31 +117,41 @@ func (x *ListStreamsResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *ListStreamsResponse) GetStream() []*Stream { +func (x *ListStreamsResponse) GetStream() *Stream { if x != nil { - if x.xxx_hidden_Stream != nil { - return *x.xxx_hidden_Stream - } + return x.xxx_hidden_Stream } return nil } -func (x *ListStreamsResponse) SetStream(v []*Stream) { - x.xxx_hidden_Stream = &v +func (x *ListStreamsResponse) SetStream(v *Stream) { + x.xxx_hidden_Stream = v +} + +func (x *ListStreamsResponse) HasStream() bool { + if x == nil { + return false + } + return x.xxx_hidden_Stream != nil +} + +func (x *ListStreamsResponse) ClearStream() { + x.xxx_hidden_Stream = nil } type ListStreamsResponse_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - // Streams is a list of event streams that can be consumed from this server. - Stream []*Stream + // Stream is an event stream that does or may contain at least one event with + // a message type ID that is among those in the [ListStreamsRequest]. + Stream *Stream } func (b0 ListStreamsResponse_builder) Build() *ListStreamsResponse { m0 := &ListStreamsResponse{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_Stream = &b.Stream + x.xxx_hidden_Stream = b.Stream return m0 } @@ -321,11 +331,8 @@ type ConsumeEventsRequest_builder struct { StreamId *uuidpb.UUID // CheckpointOffset is the offset at which the consumer begins. CheckpointOffset uint64 - // MessageTypeIds is the set of message type IDs of the events that the client - // is interested in consuming. - // - // The server MUST omit any events that do not match one of the specified - // message type IDs from responses. + // MessageTypeIds is the set of message type IDs of the event types to include + // in the responses. MessageTypeIds []*uuidpb.UUID } @@ -451,7 +458,7 @@ const file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawD "\x12ListStreamsRequest\x12>\n" + "\x10message_type_ids\x18\x01 \x03(\v2\x14.dogma.protobuf.UUIDR\x0emessageTypeIds\"S\n" + "\x13ListStreamsResponse\x12<\n" + - "\x06stream\x18\x01 \x03(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"c\n" + + "\x06stream\x18\x01 \x01(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"c\n" + "\x06Stream\x121\n" + "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x12&\n" + "\vnext_offset\x18\x02 \x01(\x04B\x05\xaa\x01\x02\b\x02R\n" + diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index 58cdc66..1a81ea7 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -18,10 +18,10 @@ service ConsumeAPI { // within an event stream. // // The server delivers every event at or after the requested checkpoint offset - // whose message type ID is among those specified in the - // [ConsumeEventsRequest]. If no such events remain in the event stream, and - // the server cannot produce new events with any of the requested message type - // IDs, it closes the gRPC stream. + // whose message type ID is among those in the [ConsumeEventsRequest]. If no + // such events remain in the event stream, and the server cannot produce new + // events with any of the requested message type IDs, it closes the gRPC + // stream. // // If the requested stream ID is not recognized a NOT_FOUND error occurs. // @@ -40,8 +40,9 @@ message ListStreamsRequest { // ListStreamsResponse is the output of the [ConsumeAPI.ListStreams] method. message ListStreamsResponse { - // Streams is a list of event streams that can be consumed from this server. - repeated Stream stream = 1; + // Stream is an event stream that does or may contain at least one event with + // a message type ID that is among those in the [ListStreamsRequest]. + Stream stream = 1; } // Stream describes an offset-based ordered event stream. @@ -64,11 +65,8 @@ message ConsumeEventsRequest { // CheckpointOffset is the offset at which the consumer begins. uint64 checkpoint_offset = 2 [features.field_presence = IMPLICIT]; - // MessageTypeIds is the set of message type IDs of the events that the client - // is interested in consuming. - // - // The server MUST omit any events that do not match one of the specified - // message type IDs from responses. + // MessageTypeIds is the set of message type IDs of the event types to include + // in the responses. repeated dogma.protobuf.UUID message_type_ids = 3; } diff --git a/grpc/eventstreamgrpc/consume_grpc.pb.go b/grpc/eventstreamgrpc/consume_grpc.pb.go index eaad355..b941cf6 100644 --- a/grpc/eventstreamgrpc/consume_grpc.pb.go +++ b/grpc/eventstreamgrpc/consume_grpc.pb.go @@ -37,10 +37,10 @@ type ConsumeAPIClient interface { // within an event stream. // // The server delivers every event at or after the requested checkpoint offset - // whose message type ID is among those specified in the - // [ConsumeEventsRequest]. If no such events remain in the event stream, and - // the server cannot produce new events with any of the requested message type - // IDs, it closes the gRPC stream. + // whose message type ID is among those in the [ConsumeEventsRequest]. If no + // such events remain in the event stream, and the server cannot produce new + // events with any of the requested message type IDs, it closes the gRPC + // stream. // // If the requested stream ID is not recognized a NOT_FOUND error occurs. // @@ -109,10 +109,10 @@ type ConsumeAPIServer interface { // within an event stream. // // The server delivers every event at or after the requested checkpoint offset - // whose message type ID is among those specified in the - // [ConsumeEventsRequest]. If no such events remain in the event stream, and - // the server cannot produce new events with any of the requested message type - // IDs, it closes the gRPC stream. + // whose message type ID is among those in the [ConsumeEventsRequest]. If no + // such events remain in the event stream, and the server cannot produce new + // events with any of the requested message type IDs, it closes the gRPC + // stream. // // If the requested stream ID is not recognized a NOT_FOUND error occurs. // diff --git a/grpc/eventstreamgrpc/consume_primo.pb.go b/grpc/eventstreamgrpc/consume_primo.pb.go index 11f4da7..f754486 100644 --- a/grpc/eventstreamgrpc/consume_primo.pb.go +++ b/grpc/eventstreamgrpc/consume_primo.pb.go @@ -69,7 +69,9 @@ func NewListStreamsResponseBuilder() *ListStreamsResponseBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *ListStreamsResponseBuilder) From(x *ListStreamsResponse) *ListStreamsResponseBuilder { proto.Reset(&b.prototype) - b.prototype.SetStream(x.GetStream()) + if x.HasStream() { + b.prototype.SetStream(x.GetStream()) + } return b } @@ -79,13 +81,15 @@ func (b *ListStreamsResponseBuilder) From(x *ListStreamsResponse) *ListStreamsRe // not modify previously constructed messages. func (b *ListStreamsResponseBuilder) Build() *ListStreamsResponse { m := &ListStreamsResponse{} - m.SetStream(b.prototype.GetStream()) + if b.prototype.HasStream() { + m.SetStream(b.prototype.GetStream()) + } return m } // WithStream configures the builder to set the Stream field to v, // then returns b. -func (b *ListStreamsResponseBuilder) WithStream(v []*Stream) *ListStreamsResponseBuilder { +func (b *ListStreamsResponseBuilder) WithStream(v *Stream) *ListStreamsResponseBuilder { b.prototype.SetStream(v) return b } From 91b98eda46f4dc5e5480618ce4d63c2bf9b40ce4 Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 09:01:03 +1000 Subject: [PATCH 08/11] Rename stuttering `StreamId()` to `Id()`. --- grpc/eventstreamgrpc/consume.pb.go | 32 ++++++++++++------------ grpc/eventstreamgrpc/consume.proto | 4 +-- grpc/eventstreamgrpc/consume_primo.pb.go | 14 +++++------ 3 files changed, 25 insertions(+), 25 deletions(-) diff --git a/grpc/eventstreamgrpc/consume.pb.go b/grpc/eventstreamgrpc/consume.pb.go index d2a9675..d0f64b6 100644 --- a/grpc/eventstreamgrpc/consume.pb.go +++ b/grpc/eventstreamgrpc/consume.pb.go @@ -158,7 +158,7 @@ func (b0 ListStreamsResponse_builder) Build() *ListStreamsResponse { // Stream describes an offset-based ordered event stream. type Stream struct { state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_StreamId *uuidpb.UUID `protobuf:"bytes,1,opt,name=stream_id,json=streamId"` + xxx_hidden_Id *uuidpb.UUID `protobuf:"bytes,1,opt,name=id"` xxx_hidden_NextOffset uint64 `protobuf:"varint,2,opt,name=next_offset,json=nextOffset"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -189,9 +189,9 @@ func (x *Stream) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *Stream) GetStreamId() *uuidpb.UUID { +func (x *Stream) GetId() *uuidpb.UUID { if x != nil { - return x.xxx_hidden_StreamId + return x.xxx_hidden_Id } return nil } @@ -203,30 +203,30 @@ func (x *Stream) GetNextOffset() uint64 { return 0 } -func (x *Stream) SetStreamId(v *uuidpb.UUID) { - x.xxx_hidden_StreamId = v +func (x *Stream) SetId(v *uuidpb.UUID) { + x.xxx_hidden_Id = v } func (x *Stream) SetNextOffset(v uint64) { x.xxx_hidden_NextOffset = v } -func (x *Stream) HasStreamId() bool { +func (x *Stream) HasId() bool { if x == nil { return false } - return x.xxx_hidden_StreamId != nil + return x.xxx_hidden_Id != nil } -func (x *Stream) ClearStreamId() { - x.xxx_hidden_StreamId = nil +func (x *Stream) ClearId() { + x.xxx_hidden_Id = nil } type Stream_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - // StreamId is a unique identifier for the stream. - StreamId *uuidpb.UUID + // Id is a unique identifier for the stream. + Id *uuidpb.UUID // NextOffset is the offset of the next event that will be written to the // stream. The first event in the stream is at offset zero; therefore, this // value is equivalent to the number of events that have been written to the @@ -238,7 +238,7 @@ func (b0 Stream_builder) Build() *Stream { m0 := &Stream{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_StreamId = b.StreamId + x.xxx_hidden_Id = b.Id x.xxx_hidden_NextOffset = b.NextOffset return m0 } @@ -458,9 +458,9 @@ const file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawD "\x12ListStreamsRequest\x12>\n" + "\x10message_type_ids\x18\x01 \x03(\v2\x14.dogma.protobuf.UUIDR\x0emessageTypeIds\"S\n" + "\x13ListStreamsResponse\x12<\n" + - "\x06stream\x18\x01 \x01(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"c\n" + - "\x06Stream\x121\n" + - "\tstream_id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\bstreamId\x12&\n" + + "\x06stream\x18\x01 \x01(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"V\n" + + "\x06Stream\x12$\n" + + "\x02id\x18\x01 \x01(\v2\x14.dogma.protobuf.UUIDR\x02id\x12&\n" + "\vnext_offset\x18\x02 \x01(\x04B\x05\xaa\x01\x02\b\x02R\n" + "nextOffset\"\xbd\x01\n" + "\x14ConsumeEventsRequest\x121\n" + @@ -488,7 +488,7 @@ var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_goType var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_depIdxs = []int32{ 5, // 0: dogma.eventstream.consume.v1.ListStreamsRequest.message_type_ids:type_name -> dogma.protobuf.UUID 2, // 1: dogma.eventstream.consume.v1.ListStreamsResponse.stream:type_name -> dogma.eventstream.consume.v1.Stream - 5, // 2: dogma.eventstream.consume.v1.Stream.stream_id:type_name -> dogma.protobuf.UUID + 5, // 2: dogma.eventstream.consume.v1.Stream.id:type_name -> dogma.protobuf.UUID 5, // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest.stream_id:type_name -> dogma.protobuf.UUID 5, // 4: dogma.eventstream.consume.v1.ConsumeEventsRequest.message_type_ids:type_name -> dogma.protobuf.UUID 6, // 5: dogma.eventstream.consume.v1.ConsumeEventsResponse.envelopes:type_name -> dogma.protobuf.MultiEnvelope diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index 1a81ea7..c52c1df 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -47,8 +47,8 @@ message ListStreamsResponse { // Stream describes an offset-based ordered event stream. message Stream { - // StreamId is a unique identifier for the stream. - dogma.protobuf.UUID stream_id = 1; + // Id is a unique identifier for the stream. + dogma.protobuf.UUID id = 1; // NextOffset is the offset of the next event that will be written to the // stream. The first event in the stream is at offset zero; therefore, this diff --git a/grpc/eventstreamgrpc/consume_primo.pb.go b/grpc/eventstreamgrpc/consume_primo.pb.go index f754486..dac1c76 100644 --- a/grpc/eventstreamgrpc/consume_primo.pb.go +++ b/grpc/eventstreamgrpc/consume_primo.pb.go @@ -110,8 +110,8 @@ func NewStreamBuilder() *StreamBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *StreamBuilder) From(x *Stream) *StreamBuilder { proto.Reset(&b.prototype) - if x.HasStreamId() { - b.prototype.SetStreamId(x.GetStreamId()) + if x.HasId() { + b.prototype.SetId(x.GetId()) } b.prototype.SetNextOffset(x.GetNextOffset()) return b @@ -123,17 +123,17 @@ func (b *StreamBuilder) From(x *Stream) *StreamBuilder { // not modify previously constructed messages. func (b *StreamBuilder) Build() *Stream { m := &Stream{} - if b.prototype.HasStreamId() { - m.SetStreamId(b.prototype.GetStreamId()) + if b.prototype.HasId() { + m.SetId(b.prototype.GetId()) } m.SetNextOffset(b.prototype.GetNextOffset()) return m } -// WithStreamId configures the builder to set the StreamId field to v, +// WithId configures the builder to set the Id field to v, // then returns b. -func (b *StreamBuilder) WithStreamId(v *uuidpb.UUID) *StreamBuilder { - b.prototype.SetStreamId(v) +func (b *StreamBuilder) WithId(v *uuidpb.UUID) *StreamBuilder { + b.prototype.SetId(v) return b } From 3a3bfb0a536647135f931f72252ac04973e9da91 Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 12:20:36 +1000 Subject: [PATCH 09/11] Add tests for MultiEnvelope.AppendBodies and TryAppendEnvelope --- protobuf/envelopepb/envelope.go | 23 +++++++ protobuf/envelopepb/envelope_test.go | 95 ++++++++++++++++++++++++++++ 2 files changed, 118 insertions(+) diff --git a/protobuf/envelopepb/envelope.go b/protobuf/envelopepb/envelope.go index 72521bd..9f71dba 100644 --- a/protobuf/envelopepb/envelope.go +++ b/protobuf/envelopepb/envelope.go @@ -5,6 +5,7 @@ import ( "fmt" "iter" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" ) @@ -83,6 +84,28 @@ func (x *MultiEnvelope) All() iter.Seq[*Envelope] { } } +// AppendBodies appends the given bodies to x. +func (x *MultiEnvelope) AppendBodies(bodies ...*Body) { + x.SetBodies( + append( + x.GetBodies(), + bodies..., + ), + ) +} + +// TryAppendEnvelope attempts to append the body from the given envelope to x. +// +// It requires that x and env share equivalent headers. If they do not, it +// returns false and does not modify x. +func (x *MultiEnvelope) TryAppendEnvelope(env *Envelope) bool { + if proto.Equal(x.GetHeader(), env.GetHeader()) { + x.AppendBodies(env.GetBody()) + return true + } + return false +} + // validate returns an error if x is not well-formed. func (x *Source) validate() error { if x == nil { diff --git a/protobuf/envelopepb/envelope_test.go b/protobuf/envelopepb/envelope_test.go index 60aaad6..ebc2334 100644 --- a/protobuf/envelopepb/envelope_test.go +++ b/protobuf/envelopepb/envelope_test.go @@ -307,6 +307,101 @@ func TestMultiEnvelope_All(t *testing.T) { }) } +func TestMultiEnvelope_AppendBodies(t *testing.T) { + t.Parallel() + + t.Run("it appends bodies to the multi-envelope", func(t *testing.T) { + t.Parallel() + + body1 := newEnvelope().GetBody() + body2 := newEnvelope().GetBody() + body3 := newEnvelope().GetBody() + + multi := NewMultiEnvelopeBuilder(). + WithHeader(newEnvelope().GetHeader()). + WithBodies([]*Body{body1}). + Build() + + multi.AppendBodies(body2, body3) + + got := multi.GetBodies() + Expect(t, "unexpected body count", len(got), 3) + Expect(t, "unexpected body[0]", got[0], body1) + Expect(t, "unexpected body[1]", got[1], body2) + Expect(t, "unexpected body[2]", got[2], body3) + }) + + t.Run("it appends to an empty multi-envelope", func(t *testing.T) { + t.Parallel() + + body := newEnvelope().GetBody() + + multi := NewMultiEnvelopeBuilder(). + WithHeader(newEnvelope().GetHeader()). + Build() + + multi.AppendBodies(body) + + got := multi.GetBodies() + Expect(t, "unexpected body count", len(got), 1) + Expect(t, "unexpected body[0]", got[0], body) + }) +} + +func TestMultiEnvelope_TryAppendEnvelope(t *testing.T) { + t.Parallel() + + t.Run("it appends when headers are equivalent", func(t *testing.T) { + t.Parallel() + + header := newEnvelope().GetHeader() + body1 := newEnvelope().GetBody() + body2 := newEnvelope().GetBody() + + multi := NewMultiEnvelopeBuilder(). + WithHeader(header). + WithBodies([]*Body{body1}). + Build() + + env := NewEnvelopeBuilder(). + WithHeader(header). + WithBody(body2). + Build() + + ok := multi.TryAppendEnvelope(env) + + Expect(t, "unexpected result", ok, true) + got := multi.GetBodies() + Expect(t, "unexpected body count", len(got), 2) + Expect(t, "unexpected body[0]", got[0], body1) + Expect(t, "unexpected body[1]", got[1], body2) + }) + + t.Run("it returns false when headers differ", func(t *testing.T) { + t.Parallel() + + body1 := newEnvelope().GetBody() + body2 := newEnvelope().GetBody() + + multi := NewMultiEnvelopeBuilder(). + WithHeader(newEnvelope().GetHeader()). + WithBodies([]*Body{body1}). + Build() + + env := NewEnvelopeBuilder(). + WithHeader(newEnvelope().GetHeader()). + WithBody(body2). + Build() + + ok := multi.TryAppendEnvelope(env) + + Expect(t, "unexpected result", ok, false) + got := multi.GetBodies() + Expect(t, "unexpected body count", len(got), 1) + Expect(t, "unexpected body[0]", got[0], body1) + }) +} + func TestWireCompatibility(t *testing.T) { t.Parallel() From a24c76c3a76f94194caba04062506b84718e24d6 Mon Sep 17 00:00:00 2001 From: James Harris Date: Thu, 9 Jul 2026 12:37:46 +1000 Subject: [PATCH 10/11] Add test for Envelope.AsMultiEnvelope --- protobuf/envelopepb/envelope.go | 12 ++++++++++++ protobuf/envelopepb/envelope_test.go | 13 +++++++++++++ 2 files changed, 25 insertions(+) diff --git a/protobuf/envelopepb/envelope.go b/protobuf/envelopepb/envelope.go index 9f71dba..154ba58 100644 --- a/protobuf/envelopepb/envelope.go +++ b/protobuf/envelopepb/envelope.go @@ -38,6 +38,18 @@ func (x *Envelope) Validate() error { return nil } +// AsMultiEnvelope returns a [MultiEnvelope] that contains the same header and +// body as x. +// +// This is a convenience method for when you want to treat a single-envelope +// message as a multi-envelope message. +func (x *Envelope) AsMultiEnvelope() *MultiEnvelope { + return NewMultiEnvelopeBuilder(). + WithHeader(x.GetHeader()). + WithBodies([]*Body{x.GetBody()}). + Build() +} + // Validate returns an error if x is not well-formed. // // Well-formedness means that all compulsory fields are populated and that no diff --git a/protobuf/envelopepb/envelope_test.go b/protobuf/envelopepb/envelope_test.go index ebc2334..b2e3f37 100644 --- a/protobuf/envelopepb/envelope_test.go +++ b/protobuf/envelopepb/envelope_test.go @@ -243,6 +243,19 @@ func TestEnvelope_Validate(t *testing.T) { }) } +func TestEnvelope_AsMultiEnvelope(t *testing.T) { + t.Parallel() + + env := newEnvelope() + multi := env.AsMultiEnvelope() + + Expect(t, "unexpected header", multi.GetHeader(), env.GetHeader()) + + bodies := multi.GetBodies() + Expect(t, "unexpected body count", len(bodies), 1) + Expect(t, "unexpected body", bodies[0], env.GetBody()) +} + func TestMultiEnvelope_All(t *testing.T) { t.Parallel() From 3e8fc5860fa39aed75f9c29a14534a47c75afccf Mon Sep 17 00:00:00 2001 From: James Harris Date: Mon, 20 Jul 2026 10:34:55 +1000 Subject: [PATCH 11/11] Remove event type IDs from list streams request. --- grpc/eventstreamgrpc/consume.pb.go | 58 ++++++++---------------- grpc/eventstreamgrpc/consume.proto | 9 +--- grpc/eventstreamgrpc/consume_grpc.pb.go | 6 +-- grpc/eventstreamgrpc/consume_primo.pb.go | 9 ---- 4 files changed, 23 insertions(+), 59 deletions(-) diff --git a/grpc/eventstreamgrpc/consume.pb.go b/grpc/eventstreamgrpc/consume.pb.go index d0f64b6..1e255a2 100644 --- a/grpc/eventstreamgrpc/consume.pb.go +++ b/grpc/eventstreamgrpc/consume.pb.go @@ -24,10 +24,9 @@ const ( // ListStreamsRequest is the input to the [ConsumeAPI.ListStreams] method. type ListStreamsRequest struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_MessageTypeIds *[]*uuidpb.UUID `protobuf:"bytes,1,rep,name=message_type_ids,json=messageTypeIds"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ListStreamsRequest) Reset() { @@ -55,32 +54,15 @@ func (x *ListStreamsRequest) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *ListStreamsRequest) GetMessageTypeIds() []*uuidpb.UUID { - if x != nil { - if x.xxx_hidden_MessageTypeIds != nil { - return *x.xxx_hidden_MessageTypeIds - } - } - return nil -} - -func (x *ListStreamsRequest) SetMessageTypeIds(v []*uuidpb.UUID) { - x.xxx_hidden_MessageTypeIds = &v -} - type ListStreamsRequest_builder struct { _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. - // MessageTypeIds is the set of message type IDs of the events that the client - // is interested in consuming. - MessageTypeIds []*uuidpb.UUID } func (b0 ListStreamsRequest_builder) Build() *ListStreamsRequest { m0 := &ListStreamsRequest{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_MessageTypeIds = &b.MessageTypeIds return m0 } @@ -454,9 +436,8 @@ var File_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto protor const file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_rawDesc = "" + "\n" + - "@github.com/dogmatiq/enginekit/grpc/eventstreamgrpc/consume.proto\x12\x1cdogma.eventstream.consume.v1\x1a@github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto\x1a8github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto\"T\n" + - "\x12ListStreamsRequest\x12>\n" + - "\x10message_type_ids\x18\x01 \x03(\v2\x14.dogma.protobuf.UUIDR\x0emessageTypeIds\"S\n" + + "@github.com/dogmatiq/enginekit/grpc/eventstreamgrpc/consume.proto\x12\x1cdogma.eventstream.consume.v1\x1a@github.com/dogmatiq/enginekit/protobuf/envelopepb/envelope.proto\x1a8github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto\"\x14\n" + + "\x12ListStreamsRequest\"S\n" + "\x13ListStreamsResponse\x12<\n" + "\x06stream\x18\x01 \x01(\v2$.dogma.eventstream.consume.v1.StreamR\x06stream\"V\n" + "\x06Stream\x12$\n" + @@ -486,21 +467,20 @@ var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_goType (*envelopepb.MultiEnvelope)(nil), // 6: dogma.protobuf.MultiEnvelope } var file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_depIdxs = []int32{ - 5, // 0: dogma.eventstream.consume.v1.ListStreamsRequest.message_type_ids:type_name -> dogma.protobuf.UUID - 2, // 1: dogma.eventstream.consume.v1.ListStreamsResponse.stream:type_name -> dogma.eventstream.consume.v1.Stream - 5, // 2: dogma.eventstream.consume.v1.Stream.id:type_name -> dogma.protobuf.UUID - 5, // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest.stream_id:type_name -> dogma.protobuf.UUID - 5, // 4: dogma.eventstream.consume.v1.ConsumeEventsRequest.message_type_ids:type_name -> dogma.protobuf.UUID - 6, // 5: dogma.eventstream.consume.v1.ConsumeEventsResponse.envelopes:type_name -> dogma.protobuf.MultiEnvelope - 0, // 6: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:input_type -> dogma.eventstream.consume.v1.ListStreamsRequest - 3, // 7: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:input_type -> dogma.eventstream.consume.v1.ConsumeEventsRequest - 1, // 8: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:output_type -> dogma.eventstream.consume.v1.ListStreamsResponse - 4, // 9: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:output_type -> dogma.eventstream.consume.v1.ConsumeEventsResponse - 8, // [8:10] is the sub-list for method output_type - 6, // [6:8] is the sub-list for method input_type - 6, // [6:6] is the sub-list for extension type_name - 6, // [6:6] is the sub-list for extension extendee - 0, // [0:6] is the sub-list for field type_name + 2, // 0: dogma.eventstream.consume.v1.ListStreamsResponse.stream:type_name -> dogma.eventstream.consume.v1.Stream + 5, // 1: dogma.eventstream.consume.v1.Stream.id:type_name -> dogma.protobuf.UUID + 5, // 2: dogma.eventstream.consume.v1.ConsumeEventsRequest.stream_id:type_name -> dogma.protobuf.UUID + 5, // 3: dogma.eventstream.consume.v1.ConsumeEventsRequest.message_type_ids:type_name -> dogma.protobuf.UUID + 6, // 4: dogma.eventstream.consume.v1.ConsumeEventsResponse.envelopes:type_name -> dogma.protobuf.MultiEnvelope + 0, // 5: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:input_type -> dogma.eventstream.consume.v1.ListStreamsRequest + 3, // 6: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:input_type -> dogma.eventstream.consume.v1.ConsumeEventsRequest + 1, // 7: dogma.eventstream.consume.v1.ConsumeAPI.ListStreams:output_type -> dogma.eventstream.consume.v1.ListStreamsResponse + 4, // 8: dogma.eventstream.consume.v1.ConsumeAPI.ConsumeEvents:output_type -> dogma.eventstream.consume.v1.ConsumeEventsResponse + 7, // [7:9] is the sub-list for method output_type + 5, // [5:7] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name } func init() { file_github_com_dogmatiq_enginekit_grpc_eventstreamgrpc_consume_proto_init() } diff --git a/grpc/eventstreamgrpc/consume.proto b/grpc/eventstreamgrpc/consume.proto index c52c1df..6ab51b5 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -10,8 +10,7 @@ import "github.com/dogmatiq/enginekit/protobuf/uuidpb/uuid.proto"; // ConsumeAPI is a service for consuming event messages from offset-based, // ordered event streams. service ConsumeAPI { - // ListStreams returns the event streams that may contain events whose - // message type IDs are among those specified in the [ListStreamsRequest]. + // ListStreams returns the event streams offered by the server. rpc ListStreams(ListStreamsRequest) returns (stream ListStreamsResponse); // ConsumeEvents returns events in order, starting from a specific offset @@ -32,11 +31,7 @@ service ConsumeAPI { } // ListStreamsRequest is the input to the [ConsumeAPI.ListStreams] method. -message ListStreamsRequest { - // MessageTypeIds is the set of message type IDs of the events that the client - // is interested in consuming. - repeated dogma.protobuf.UUID message_type_ids = 1; -} +message ListStreamsRequest {} // ListStreamsResponse is the output of the [ConsumeAPI.ListStreams] method. message ListStreamsResponse { diff --git a/grpc/eventstreamgrpc/consume_grpc.pb.go b/grpc/eventstreamgrpc/consume_grpc.pb.go index b941cf6..4e21782 100644 --- a/grpc/eventstreamgrpc/consume_grpc.pb.go +++ b/grpc/eventstreamgrpc/consume_grpc.pb.go @@ -30,8 +30,7 @@ const ( // ConsumeAPI is a service for consuming event messages from offset-based, // ordered event streams. type ConsumeAPIClient interface { - // ListStreams returns the event streams that may contain events whose - // message type IDs are among those specified in the [ListStreamsRequest]. + // ListStreams returns the event streams offered by the server. ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListStreamsResponse], error) // ConsumeEvents returns events in order, starting from a specific offset // within an event stream. @@ -102,8 +101,7 @@ type ConsumeAPI_ConsumeEventsClient = grpc.ServerStreamingClient[ConsumeEventsRe // ConsumeAPI is a service for consuming event messages from offset-based, // ordered event streams. type ConsumeAPIServer interface { - // ListStreams returns the event streams that may contain events whose - // message type IDs are among those specified in the [ListStreamsRequest]. + // ListStreams returns the event streams offered by the server. ListStreams(*ListStreamsRequest, grpc.ServerStreamingServer[ListStreamsResponse]) error // ConsumeEvents returns events in order, starting from a specific offset // within an event stream. diff --git a/grpc/eventstreamgrpc/consume_primo.pb.go b/grpc/eventstreamgrpc/consume_primo.pb.go index dac1c76..23b0520 100644 --- a/grpc/eventstreamgrpc/consume_primo.pb.go +++ b/grpc/eventstreamgrpc/consume_primo.pb.go @@ -32,7 +32,6 @@ func NewListStreamsRequestBuilder() *ListStreamsRequestBuilder { // do not modify x. It does not make a copy of the field values themselves. func (b *ListStreamsRequestBuilder) From(x *ListStreamsRequest) *ListStreamsRequestBuilder { proto.Reset(&b.prototype) - b.prototype.SetMessageTypeIds(x.GetMessageTypeIds()) return b } @@ -42,17 +41,9 @@ func (b *ListStreamsRequestBuilder) From(x *ListStreamsRequest) *ListStreamsRequ // not modify previously constructed messages. func (b *ListStreamsRequestBuilder) Build() *ListStreamsRequest { m := &ListStreamsRequest{} - m.SetMessageTypeIds(b.prototype.GetMessageTypeIds()) return m } -// WithMessageTypeIds configures the builder to set the MessageTypeIds field to v, -// then returns b. -func (b *ListStreamsRequestBuilder) WithMessageTypeIds(v []*uuidpb.UUID) *ListStreamsRequestBuilder { - b.prototype.SetMessageTypeIds(v) - return b -} - type ListStreamsResponseBuilder struct { prototype ListStreamsResponse }