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.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 d71c26d..1e255a2 100644 --- a/grpc/eventstreamgrpc/consume.pb.go +++ b/grpc/eventstreamgrpc/consume.pb.go @@ -22,7 +22,7 @@ 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 @@ -66,12 +66,12 @@ func (b0 ListStreamsRequest_builder) Build() *ListStreamsRequest { 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_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,41 +99,51 @@ 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 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_Streams = &b.Streams + 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_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 } func (x *Stream) Reset() { @@ -161,67 +171,68 @@ 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 } -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) SetId(v *uuidpb.UUID) { + x.xxx_hidden_Id = 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 { +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 - // EventTypeIds is the set of type IDs of events that may appear on the stream. - EventTypeIds []*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 + // stream. + NextOffset uint64 } func (b0 Stream_builder) Build() *Stream { m0 := &Stream{} b, x := &b0, m0 _, _ = b, x - x.xxx_hidden_StreamId = b.StreamId - x.xxx_hidden_EventTypeIds = &b.EventTypeIds + x.xxx_hidden_Id = b.Id + 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() { @@ -256,17 +267,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 @@ -276,12 +287,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 { @@ -300,12 +311,11 @@ 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 event types to include + // in the responses. + MessageTypeIds []*uuidpb.UUID } func (b0 ConsumeEventsRequest_builder) Build() *ConsumeEventsRequest { @@ -313,18 +323,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() { @@ -352,371 +374,61 @@ func (x *ConsumeEventsResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -func (x *ConsumeEventsResponse) GetEventDelivery() *ConsumeEventsResponse_EventDelivery { - 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] +func (x *ConsumeEventsResponse) GetCheckpointOffset() uint64 { 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 } @@ -725,65 +437,50 @@ 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" + - "\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" + + "\x12ListStreamsRequest\"S\n" + + "\x13ListStreamsResponse\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" + - "\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\xfc\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\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) +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.streams: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 + 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() } @@ -791,16 +488,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 451c260..6ab51b5 100644 --- a/grpc/eventstreamgrpc/consume.proto +++ b/grpc/eventstreamgrpc/consume.proto @@ -7,95 +7,97 @@ 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 offset-based, +// ordered event streams. service ConsumeAPI { - // ListStreams lists the streams that the server provides. - rpc ListStreams(ListStreamsRequest) returns (ListStreamsResponse); + // ListStreams returns the event streams offered by the server. + rpc ListStreams(ListStreamsRequest) returns (stream ListStreamsResponse); - // Consume 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 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. +// ListStreamsRequest is the input to the [ConsumeAPI.ListStreams] method. message ListStreamsRequest {} -// ListStreamsResponse is the output of the ConsumeAPI.ListStreams method. +// 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 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. 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; + // 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 + // 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 event types to include + // in the 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 d53eb7e..4e21782 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 @@ -27,28 +27,24 @@ 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 offset-based, +// ordered event streams. type ConsumeAPIClient interface { - // ListStreams lists the streams that the server provides. - ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (*ListStreamsResponse, error) - // Consume starts consuming from a specific offset within an event stream. + // 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. // - // 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 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. ConsumeEvents(ctx context.Context, in *ConsumeEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ConsumeEventsResponse], error) } @@ -60,19 +56,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 } @@ -93,28 +98,24 @@ 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 offset-based, +// ordered event streams. type ConsumeAPIServer interface { - // ListStreams lists the streams that the server provides. - ListStreams(context.Context, *ListStreamsRequest) (*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]. + // 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. // - // 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 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 } @@ -125,8 +126,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 +152,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 +180,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..23b0520 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 } @@ -69,7 +60,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 +72,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 } @@ -106,10 +101,10 @@ 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.SetEventTypeIds(x.GetEventTypeIds()) + b.prototype.SetNextOffset(x.GetNextOffset()) return b } @@ -119,24 +114,24 @@ 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.SetEventTypeIds(b.prototype.GetEventTypeIds()) + 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 } -// 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 } @@ -159,8 +154,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 } @@ -173,8 +168,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 } @@ -185,17 +180,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 } @@ -215,9 +210,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 } @@ -227,260 +221,38 @@ 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()) - } + m.SetCheckpointOffset(b.prototype.GetCheckpointOffset()) + m.SetEnvelopes(b.prototype.GetEnvelopes()) return m } -// WithEventDelivery configures the builder to set the EventDelivery field to v, +// WithCheckpointOffset configures the builder to set the CheckpointOffset field to v, // then returns b. -func (b *ConsumeEventsResponseBuilder) WithEventDelivery(v *ConsumeEventsResponse_EventDelivery) *ConsumeEventsResponseBuilder { - b.prototype.SetEventDelivery(v) +func (b *ConsumeEventsResponseBuilder) WithCheckpointOffset(v uint64) *ConsumeEventsResponseBuilder { + b.prototype.SetCheckpointOffset(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, +// 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 *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()) - } +func (b *ConsumeEventsResponseBuilder) WithEnvelopes(v []*envelopepb.MultiEnvelope) *ConsumeEventsResponseBuilder { + b.prototype.SetEnvelopes(v) 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()) - } - return m -} - -// WithStreamId configures the builder to set the StreamId field to v, -// then returns b. -func (b *UnrecognizedStreamBuilder) WithStreamId(v *uuidpb.UUID) *UnrecognizedStreamBuilder { - b.prototype.SetStreamId(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, -// 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) - 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) 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") } @@ -574,67 +346,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) -} diff --git a/protobuf/envelopepb/envelope.go b/protobuf/envelopepb/envelope.go index 72521bd..154ba58 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" ) @@ -37,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 @@ -83,6 +96,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..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() @@ -307,6 +320,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()