// Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 // - protoc v7.34.1 // source: waymaker_streams.proto package waymaker_streams import ( context "context" grpc "google.golang.org/grpc" codes "google.golang.org/grpc/codes" status "google.golang.org/grpc/status" ) // This is a compile-time assertion to ensure that this generated file // is compatible with the grpc package it is being compiled against. // Requires gRPC-Go v1.64.0 or later. const _ = grpc.SupportPackageIsVersion9 const ( WaymakerStreamsService_CreateStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/CreateStream" WaymakerStreamsService_DeleteStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/DeleteStream" WaymakerStreamsService_GetStreamInfo_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetStreamInfo" WaymakerStreamsService_ListStreams_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ListStreams" WaymakerStreamsService_GetStreamSources_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetStreamSources" WaymakerStreamsService_UpdateStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/UpdateStream" WaymakerStreamsService_Publish_FullMethodName = "/waymaker.streams.WaymakerStreamsService/Publish" WaymakerStreamsService_Fetch_FullMethodName = "/waymaker.streams.WaymakerStreamsService/Fetch" WaymakerStreamsService_Ack_FullMethodName = "/waymaker.streams.WaymakerStreamsService/Ack" WaymakerStreamsService_Nak_FullMethodName = "/waymaker.streams.WaymakerStreamsService/Nak" WaymakerStreamsService_Term_FullMethodName = "/waymaker.streams.WaymakerStreamsService/Term" WaymakerStreamsService_InProgress_FullMethodName = "/waymaker.streams.WaymakerStreamsService/InProgress" WaymakerStreamsService_Subscribe_FullMethodName = "/waymaker.streams.WaymakerStreamsService/Subscribe" WaymakerStreamsService_CreateConsumer_FullMethodName = "/waymaker.streams.WaymakerStreamsService/CreateConsumer" WaymakerStreamsService_DeleteConsumer_FullMethodName = "/waymaker.streams.WaymakerStreamsService/DeleteConsumer" WaymakerStreamsService_ListConsumers_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ListConsumers" WaymakerStreamsService_GetConsumerInfo_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetConsumerInfo" WaymakerStreamsService_TransferStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/TransferStream" WaymakerStreamsService_MigrateStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/MigrateStream" WaymakerStreamsService_GetClusterStreamStats_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetClusterStreamStats" WaymakerStreamsService_WatchStreams_FullMethodName = "/waymaker.streams.WaymakerStreamsService/WatchStreams" WaymakerStreamsService_ReadLatestAtSubject_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReadLatestAtSubject" WaymakerStreamsService_ListSubjectsByPrefix_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ListSubjectsByPrefix" WaymakerStreamsService_ScanExactAtSubject_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ScanExactAtSubject" WaymakerStreamsService_ClearStreamAuthority_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ClearStreamAuthority" WaymakerStreamsService_ListStreamAuthorityOverrides_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ListStreamAuthorityOverrides" WaymakerStreamsService_SetStreamPinned_FullMethodName = "/waymaker.streams.WaymakerStreamsService/SetStreamPinned" WaymakerStreamsService_PutObject_FullMethodName = "/waymaker.streams.WaymakerStreamsService/PutObject" WaymakerStreamsService_GetObject_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetObject" WaymakerStreamsService_DeleteObject_FullMethodName = "/waymaker.streams.WaymakerStreamsService/DeleteObject" WaymakerStreamsService_GetObjectInfo_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetObjectInfo" WaymakerStreamsService_ListObjects_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ListObjects" WaymakerStreamsService_PutObjectStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/PutObjectStream" WaymakerStreamsService_GetObjectStream_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetObjectStream" WaymakerStreamsService_ListObjectRevisions_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ListObjectRevisions" WaymakerStreamsService_GetObjectRange_FullMethodName = "/waymaker.streams.WaymakerStreamsService/GetObjectRange" WaymakerStreamsService_RebalanceStreams_FullMethodName = "/waymaker.streams.WaymakerStreamsService/RebalanceStreams" WaymakerStreamsService_ReplicateConsumerState_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateConsumerState" WaymakerStreamsService_ReplicateSourceTailState_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateSourceTailState" WaymakerStreamsService_ReplicateStreamCreate_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateStreamCreate" WaymakerStreamsService_ReplicateMessage_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateMessage" WaymakerStreamsService_ReplicateStreamDelete_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateStreamDelete" WaymakerStreamsService_ReplicateTruncate_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateTruncate" WaymakerStreamsService_ReplicateStreamUpdate_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateStreamUpdate" WaymakerStreamsService_ReplicateWorkQueueAck_FullMethodName = "/waymaker.streams.WaymakerStreamsService/ReplicateWorkQueueAck" ) // WaymakerStreamsServiceClient is the client API for WaymakerStreamsService service. // // 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. type WaymakerStreamsServiceClient interface { // --- Stream lifecycle --- CreateStream(ctx context.Context, in *CreateStreamRequest, opts ...grpc.CallOption) (*CreateStreamResponse, error) DeleteStream(ctx context.Context, in *DeleteStreamRequest, opts ...grpc.CallOption) (*DeleteStreamResponse, error) GetStreamInfo(ctx context.Context, in *GetStreamInfoRequest, opts ...grpc.CallOption) (*GetStreamInfoResponse, error) ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (*ListStreamsResponse, error) // Slice 3 cross-stream sources admin: enumerate every // (sourcing, source) tail running on this node, with current // last_sourced_seq + pulled_total + last_error. Useful for // operators auditing the cluster's source topology without // ListStreams + GetStreamInfo per stream. GetStreamSources(ctx context.Context, in *GetStreamSourcesRequest, opts ...grpc.CallOption) (*GetStreamSourcesResponse, error) // Update the *mutable* subset of a stream's config — the Limits // retention bounds (max_age_ms / max_msgs / max_bytes), the per- // message size cap, and the strict-limits toggle. Immutable fields // (name, subjects_filter, block_size, retention policy type) are // not touched. Lowering a bound triggers an immediate prune to // bring stats under the new limit; the primary fans the resulting // truncation out via `ReplicateTruncate` so secondaries mirror. // Partial-update semantics: only fields explicitly set in the // request are applied; unset fields leave the on-disk value // unchanged. UpdateStream(ctx context.Context, in *UpdateStreamRequest, opts ...grpc.CallOption) (*UpdateStreamResponse, error) // --- Messages --- Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*PublishResponse, error) Fetch(ctx context.Context, in *FetchRequest, opts ...grpc.CallOption) (*FetchResponse, error) Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error) // Negative-acknowledge: server resets the pending entry's // delivered_at_ms so the next fetch redelivers. `delay_ms` defers // eligibility by that wall-clock window (0 = immediate). The // message's `deliver_count` keeps climbing toward `max_deliver`. Nak(ctx context.Context, in *NakRequest, opts ...grpc.CallOption) (*NakResponse, error) // Terminal-acknowledge: drop the pending entry permanently // without redelivery, regardless of `max_deliver`. Does NOT // trigger WorkQueue delete — other consumers can still observe // the message. Term(ctx context.Context, in *TermRequest, opts ...grpc.CallOption) (*TermResponse, error) // Heartbeat-acknowledge: bump delivered_at_ms = now to extend // the ack_wait window. `deliver_count` is unchanged. InProgress(ctx context.Context, in *InProgressRequest, opts ...grpc.CallOption) (*InProgressResponse, error) // Push-mode delivery: the server fetches in a loop and streams // each delivered message back to the client as it arrives. The // client acks via the unary Ack RPC just like pull-mode. The // stream stays open until the client disconnects, the server // returns an error, or the consumer is deleted. Wakes // immediately on new appends via the storage layer's subscribe // primitive — no polling for empty streams. Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SubscribeEvent], error) // --- Consumers --- CreateConsumer(ctx context.Context, in *CreateConsumerRequest, opts ...grpc.CallOption) (*CreateConsumerResponse, error) DeleteConsumer(ctx context.Context, in *DeleteConsumerRequest, opts ...grpc.CallOption) (*DeleteConsumerResponse, error) ListConsumers(ctx context.Context, in *ListConsumersRequest, opts ...grpc.CallOption) (*ListConsumersResponse, error) GetConsumerInfo(ctx context.Context, in *GetConsumerInfoRequest, opts ...grpc.CallOption) (*GetConsumerInfoResponse, error) // --- Rebalancing (Phase 1: operator-driven only) --- // // The current owner of a stream serves its raw redb bytes to a peer // that's pulling the stream over. The handler atomically removes the // stream from its local registry first, refusing the call if any // outside reference is still live (operator must drain writers). On // RPC success the source deletes the local file. See // STREAMS_SPEC.md §11 for the model and limitations (no automatic // ring-change sweep yet; the operator is responsible for triggering // a migrate when membership moves a stream's authority). TransferStream(ctx context.Context, in *TransferStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[TransferStreamChunk], error) // Admin trigger on the receiving side: pull stream `name` from // `source_node_id`'s `TransferStream` and own it locally. MigrateStream(ctx context.Context, in *MigrateStreamRequest, opts ...grpc.CallOption) (*MigrateStreamResponse, error) // Cluster-wide stream inventory + skew report. The receiving node // queries every cluster member's local `StreamsRegistry` (via the // existing proxy channel pool) and aggregates the result. Used by // operators to identify hash-skew imbalance before triggering // `RebalanceStreams`. Also exposed via the `wmkr-status` CLI. GetClusterStreamStats(ctx context.Context, in *GetClusterStreamStatsRequest, opts ...grpc.CallOption) (*GetClusterStreamStatsResponse, error) // Server-streamed admin watch — emits a WatchEvent each time the // local node's state mutates (stream / consumer create / delete / // update). Useful for live dashboards or service-discovery // clients that want to react to topology changes without // polling. Local-only for now: each watcher sees events generated // on the node it connected to. Cluster-wide watch can be built // on top via a fan-out client; the server doesn't fan out // automatically because the events would arrive out of any // single-source ordering anyway under proxy hops. WatchStreams(ctx context.Context, in *WatchStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[WatchEvent], error) // Read the latest message at a given subject within a stream. // The foundation for KV-style "last-value wins" lookups on top // of a stream — KV put = Publish to `.`; KV get = // this RPC against the same subject. Returns the full // MessagePb (including headers) so callers can detect KV // tombstones (`wmkv.tombstone` header). // // Returns `success: true` with `message` unset when no message // has ever been published at this subject (or all have been // pruned). Routes via `try_route!` like every other per-stream // RPC. ReadLatestAtSubject(ctx context.Context, in *ReadLatestAtSubjectRequest, opts ...grpc.CallOption) (*ReadLatestAtSubjectResponse, error) // List every distinct subject in `stream` whose name starts // with `prefix`. Cost is O(matching subjects); independent of // message count. The foundation for `streams-cli kv-keys` and // service-discovery-style "everything under this namespace" // lookups. Returns subjects whose latest message is a // tombstone too — clients that want live-keys-only filter // tombstones via a follow-up `ReadLatestAtSubject`. ListSubjectsByPrefix(ctx context.Context, in *ListSubjectsByPrefixRequest, opts ...grpc.CallOption) (*ListSubjectsByPrefixResponse, error) // Scan all messages published at an exact subject within // `stream`, in seq order, starting at `from_seq` (0 = from the // beginning), bounded by `limit`. The foundation for // `streams-cli kv-history` — operators want to inspect every // value ever published under a KV key (including tombstones) // for debugging/audit. Cost is O(matching messages); independent // of total stream size. Routes via `try_route!` like every // other per-stream RPC. ScanExactAtSubject(ctx context.Context, in *ScanExactAtSubjectRequest, opts ...grpc.CallOption) (*ScanExactAtSubjectResponse, error) // Remove a Phase 3 per-stream authority override. Routing // reverts to the ring's hash owner. Idempotent: clearing a // stream with no override succeeds silently. Operators use this // to retire a stale override (e.g. after a ring shift made the // override redundant). Commits via a Raft entry so the clear // applies on every node before the response returns. ClearStreamAuthority(ctx context.Context, in *ClearStreamAuthorityRequest, opts ...grpc.CallOption) (*ClearStreamAuthorityResponse, error) // List every Phase 3 stream_authority override active on the // responding node. The map is Raft-replicated, so any node's // response reflects the cluster-wide view (modulo apply lag). // Useful for ops triage when an unexpected number of overrides // shows up on /metrics. No fan-out — single-node RPC; the // returned set is the canonical truth. ListStreamAuthorityOverrides(ctx context.Context, in *ListStreamAuthorityOverridesRequest, opts ...grpc.CallOption) (*ListStreamAuthorityOverridesResponse, error) // Toggle pinned state for `stream`. Pinned streams are exempt // from the auto-GC sweep that retires redundant overrides — use // when you want a stream to stay on its current authority node // even if the ring shifts to make the override redundant. // Idempotent. Independent of the override itself (pinning a // stream with no override is benign; the marker sits dormant). SetStreamPinned(ctx context.Context, in *SetStreamPinnedRequest, opts ...grpc.CallOption) (*SetStreamPinnedResponse, error) PutObject(ctx context.Context, in *PutObjectRequest, opts ...grpc.CallOption) (*PutObjectResponse, error) GetObject(ctx context.Context, in *GetObjectRequest, opts ...grpc.CallOption) (*GetObjectResponse, error) DeleteObject(ctx context.Context, in *DeleteObjectRequest, opts ...grpc.CallOption) (*DeleteObjectResponse, error) GetObjectInfo(ctx context.Context, in *GetObjectInfoRequest, opts ...grpc.CallOption) (*GetObjectInfoResponse, error) ListObjects(ctx context.Context, in *ListObjectsRequest, opts ...grpc.CallOption) (*ListObjectsResponse, error) // Client-streamed PutObject for arbitrary-size objects. First // frame MUST set `start { bucket, name, chunk_size, headers, // sha256 }`. Subsequent frames carry `data` only — each frame's // `data` is ONE chunk message at `objc..`. The server // accumulates a running SHA-256 and total-byte count, publishes // chunks as they arrive (replication fires async), and on the // last frame (`finish=true`) publishes the metadata. A client // disconnect before `finish=true` leaves orphan chunks; the GC // sweep cleans them up. PutObjectStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutObjectStreamFrame, PutObjectResponse], error) // Server-streamed GetObject. First frame carries `info`; // subsequent frames carry `data` only — one per chunk. Last // frame sets `done=true`. The client reassembles; the response // is sent over the wire in chunk-sized pieces so memory usage // stays bounded on both sides. GetObjectStream(ctx context.Context, in *GetObjectRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetObjectStreamFrame], error) // Every revision of `name`'s metadata in seq order — covers // overwrites + tombstones. Returns one entry per metadata // message at `objm.`. Chunks are not enumerated; this RPC // is for object versioning / audit, not for binary diffing. ListObjectRevisions(ctx context.Context, in *ListObjectRevisionsRequest, opts ...grpc.CallOption) (*ListObjectRevisionsResponse, error) // Read a byte range `[offset, offset + len)` from an object's // assembled payload. Only the chunks that intersect the range // are loaded server-side — useful for resumable downloads of // large objects. // - `offset + len > total_bytes` → returns whatever bytes exist // in the range (success, possibly empty). // - `offset > total_bytes` → returns empty payload (success). // - `len == 0` → returns empty payload (success). GetObjectRange(ctx context.Context, in *GetObjectRangeRequest, opts ...grpc.CallOption) (*GetObjectRangeResponse, error) // Operator-driven rebalance. Takes an explicit plan — a list of // (stream, target_node) — and executes each step by issuing a // `MigrateStream` to the target. The plan is *not* auto-generated; // the operator (or a future automatic planner) is responsible for // building it from a `GetClusterStreamStats` snapshot. Steps run // sequentially with a per-step timeout; the response carries // per-step outcomes so partial success is visible. RebalanceStreams(ctx context.Context, in *RebalanceStreamsRequest, opts ...grpc.CallOption) (*RebalanceStreamsResponse, error) // --- Consumer-state replication (Phase 2 §G) --- // // The primary for a stream pushes its consumers' full state to the // stream's `replication_factor - 1` secondaries after every // state-mutating consumer operation (create_consumer, fetch, ack, // delete_consumer). The push is fire-and-forget on the primary's // side — the client RPC has already returned to the caller; the // replication runs in a background task. Secondaries hold the // snapshot in memory; adoption-on-failover is a future slice. ReplicateConsumerState(ctx context.Context, in *ReplicateConsumerStateRequest, opts ...grpc.CallOption) (*ReplicateConsumerStateResponse, error) // --- Cross-stream sources state replication (slice 2E) --- // // The primary for a sourcing stream pushes the current per-source // tail watermark to each secondary after every successful batch // (i.e. once per ~128 source messages). Secondaries persist the // snapshot via their own SourceTailStore so that on adoption (ring // shift → secondary becomes primary), `spawn_source_tail_tasks` // reads the replicated state and resumes from `last_sourced_seq + 1` // instead of re-pulling from `start_seq` (which would emit // duplicates with already-replicated provenance headers). ReplicateSourceTailState(ctx context.Context, in *ReplicateSourceTailStateRequest, opts ...grpc.CallOption) (*ReplicateSourceTailStateResponse, error) // --- Stream-data replication (Phase 3, chunk 1) --- // // The primary for a stream pushes: // 1. ReplicateStreamCreate once at create time, so secondaries // know what stream to open in their replica registry with // what config (block_size, retention, max_msg_bytes, etc.). // 2. ReplicateMessage on every successful Publish, with the // seq the primary assigned, so the secondary's replica // mirrors the message log by seq exactly. // // Replica streams live in a per-node "replica registry" rooted at // `/replicas/.redb`, distinct from the // primary-owned namespace. The streams handler never serves // client requests from the replica — it's purely catastrophe // recovery state until the (future) adoption-on-failover slice // promotes a replica to primary. ReplicateStreamCreate(ctx context.Context, in *ReplicateStreamCreateRequest, opts ...grpc.CallOption) (*ReplicateStreamCreateResponse, error) ReplicateMessage(ctx context.Context, in *ReplicateMessageRequest, opts ...grpc.CallOption) (*ReplicateMessageResponse, error) // Tear down the replica when the primary deletes the stream. // Idempotent — missing replica is success. ReplicateStreamDelete(ctx context.Context, in *ReplicateStreamDeleteRequest, opts ...grpc.CallOption) (*ReplicateStreamDeleteResponse, error) // The primary's retention sweep removed messages below // `first_seq`; the secondary mirrors the same truncation so its // replica's first_seq advances in lockstep. Idempotent. ReplicateTruncate(ctx context.Context, in *ReplicateTruncateRequest, opts ...grpc.CallOption) (*ReplicateTruncateResponse, error) // The primary applied an UpdateStream; secondaries mirror the // mutable subset of the config so a future failover lands on a // replica whose retention matches the primary's. Carries the same // narrow shape as UpdateStreamRequest — only the mutable fields, // with partial-update semantics. ReplicateStreamUpdate(ctx context.Context, in *ReplicateStreamUpdateRequest, opts ...grpc.CallOption) (*ReplicateStreamUpdateResponse, error) // Under `RetentionPolicy::WorkQueue` the primary deletes a message // on ack (delete-on-first-ack). Without this fan-out, secondaries' // replica files would still hold the acked message — and after a // failover, a fresh consumer on the new primary would see it and // re-deliver, breaking the "each message belongs to exactly one // consumer at a time" invariant. Idempotent: missing seq on // secondary is success. ReplicateWorkQueueAck(ctx context.Context, in *ReplicateWorkQueueAckRequest, opts ...grpc.CallOption) (*ReplicateWorkQueueAckResponse, error) } type waymakerStreamsServiceClient struct { cc grpc.ClientConnInterface } func NewWaymakerStreamsServiceClient(cc grpc.ClientConnInterface) WaymakerStreamsServiceClient { return &waymakerStreamsServiceClient{cc} } func (c *waymakerStreamsServiceClient) CreateStream(ctx context.Context, in *CreateStreamRequest, opts ...grpc.CallOption) (*CreateStreamResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(CreateStreamResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_CreateStream_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) DeleteStream(ctx context.Context, in *DeleteStreamRequest, opts ...grpc.CallOption) (*DeleteStreamResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(DeleteStreamResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_DeleteStream_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetStreamInfo(ctx context.Context, in *GetStreamInfoRequest, opts ...grpc.CallOption) (*GetStreamInfoResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetStreamInfoResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetStreamInfo_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ListStreams(ctx context.Context, in *ListStreamsRequest, opts ...grpc.CallOption) (*ListStreamsResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListStreamsResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ListStreams_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetStreamSources(ctx context.Context, in *GetStreamSourcesRequest, opts ...grpc.CallOption) (*GetStreamSourcesResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetStreamSourcesResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetStreamSources_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) UpdateStream(ctx context.Context, in *UpdateStreamRequest, opts ...grpc.CallOption) (*UpdateStreamResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(UpdateStreamResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_UpdateStream_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*PublishResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(PublishResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_Publish_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) Fetch(ctx context.Context, in *FetchRequest, opts ...grpc.CallOption) (*FetchResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(FetchResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_Fetch_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) Ack(ctx context.Context, in *AckRequest, opts ...grpc.CallOption) (*AckResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(AckResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_Ack_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) Nak(ctx context.Context, in *NakRequest, opts ...grpc.CallOption) (*NakResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(NakResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_Nak_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) Term(ctx context.Context, in *TermRequest, opts ...grpc.CallOption) (*TermResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(TermResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_Term_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) InProgress(ctx context.Context, in *InProgressRequest, opts ...grpc.CallOption) (*InProgressResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(InProgressResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_InProgress_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SubscribeEvent], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WaymakerStreamsService_ServiceDesc.Streams[0], WaymakerStreamsService_Subscribe_FullMethodName, cOpts...) if err != nil { return nil, err } x := &grpc.GenericClientStream[SubscribeRequest, SubscribeEvent]{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 WaymakerStreamsService_SubscribeClient = grpc.ServerStreamingClient[SubscribeEvent] func (c *waymakerStreamsServiceClient) CreateConsumer(ctx context.Context, in *CreateConsumerRequest, opts ...grpc.CallOption) (*CreateConsumerResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(CreateConsumerResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_CreateConsumer_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) DeleteConsumer(ctx context.Context, in *DeleteConsumerRequest, opts ...grpc.CallOption) (*DeleteConsumerResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(DeleteConsumerResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_DeleteConsumer_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ListConsumers(ctx context.Context, in *ListConsumersRequest, opts ...grpc.CallOption) (*ListConsumersResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListConsumersResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ListConsumers_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetConsumerInfo(ctx context.Context, in *GetConsumerInfoRequest, opts ...grpc.CallOption) (*GetConsumerInfoResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetConsumerInfoResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetConsumerInfo_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) TransferStream(ctx context.Context, in *TransferStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[TransferStreamChunk], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WaymakerStreamsService_ServiceDesc.Streams[1], WaymakerStreamsService_TransferStream_FullMethodName, cOpts...) if err != nil { return nil, err } x := &grpc.GenericClientStream[TransferStreamRequest, TransferStreamChunk]{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 WaymakerStreamsService_TransferStreamClient = grpc.ServerStreamingClient[TransferStreamChunk] func (c *waymakerStreamsServiceClient) MigrateStream(ctx context.Context, in *MigrateStreamRequest, opts ...grpc.CallOption) (*MigrateStreamResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(MigrateStreamResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_MigrateStream_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetClusterStreamStats(ctx context.Context, in *GetClusterStreamStatsRequest, opts ...grpc.CallOption) (*GetClusterStreamStatsResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetClusterStreamStatsResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetClusterStreamStats_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) WatchStreams(ctx context.Context, in *WatchStreamsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[WatchEvent], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WaymakerStreamsService_ServiceDesc.Streams[2], WaymakerStreamsService_WatchStreams_FullMethodName, cOpts...) if err != nil { return nil, err } x := &grpc.GenericClientStream[WatchStreamsRequest, WatchEvent]{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 WaymakerStreamsService_WatchStreamsClient = grpc.ServerStreamingClient[WatchEvent] func (c *waymakerStreamsServiceClient) ReadLatestAtSubject(ctx context.Context, in *ReadLatestAtSubjectRequest, opts ...grpc.CallOption) (*ReadLatestAtSubjectResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReadLatestAtSubjectResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReadLatestAtSubject_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ListSubjectsByPrefix(ctx context.Context, in *ListSubjectsByPrefixRequest, opts ...grpc.CallOption) (*ListSubjectsByPrefixResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListSubjectsByPrefixResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ListSubjectsByPrefix_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ScanExactAtSubject(ctx context.Context, in *ScanExactAtSubjectRequest, opts ...grpc.CallOption) (*ScanExactAtSubjectResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ScanExactAtSubjectResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ScanExactAtSubject_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ClearStreamAuthority(ctx context.Context, in *ClearStreamAuthorityRequest, opts ...grpc.CallOption) (*ClearStreamAuthorityResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ClearStreamAuthorityResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ClearStreamAuthority_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ListStreamAuthorityOverrides(ctx context.Context, in *ListStreamAuthorityOverridesRequest, opts ...grpc.CallOption) (*ListStreamAuthorityOverridesResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListStreamAuthorityOverridesResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ListStreamAuthorityOverrides_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) SetStreamPinned(ctx context.Context, in *SetStreamPinnedRequest, opts ...grpc.CallOption) (*SetStreamPinnedResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(SetStreamPinnedResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_SetStreamPinned_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) PutObject(ctx context.Context, in *PutObjectRequest, opts ...grpc.CallOption) (*PutObjectResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(PutObjectResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_PutObject_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetObject(ctx context.Context, in *GetObjectRequest, opts ...grpc.CallOption) (*GetObjectResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetObjectResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetObject_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) DeleteObject(ctx context.Context, in *DeleteObjectRequest, opts ...grpc.CallOption) (*DeleteObjectResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(DeleteObjectResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_DeleteObject_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetObjectInfo(ctx context.Context, in *GetObjectInfoRequest, opts ...grpc.CallOption) (*GetObjectInfoResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetObjectInfoResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetObjectInfo_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ListObjects(ctx context.Context, in *ListObjectsRequest, opts ...grpc.CallOption) (*ListObjectsResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListObjectsResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ListObjects_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) PutObjectStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[PutObjectStreamFrame, PutObjectResponse], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WaymakerStreamsService_ServiceDesc.Streams[3], WaymakerStreamsService_PutObjectStream_FullMethodName, cOpts...) if err != nil { return nil, err } x := &grpc.GenericClientStream[PutObjectStreamFrame, PutObjectResponse]{ClientStream: stream} 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 WaymakerStreamsService_PutObjectStreamClient = grpc.ClientStreamingClient[PutObjectStreamFrame, PutObjectResponse] func (c *waymakerStreamsServiceClient) GetObjectStream(ctx context.Context, in *GetObjectRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[GetObjectStreamFrame], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &WaymakerStreamsService_ServiceDesc.Streams[4], WaymakerStreamsService_GetObjectStream_FullMethodName, cOpts...) if err != nil { return nil, err } x := &grpc.GenericClientStream[GetObjectRequest, GetObjectStreamFrame]{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 WaymakerStreamsService_GetObjectStreamClient = grpc.ServerStreamingClient[GetObjectStreamFrame] func (c *waymakerStreamsServiceClient) ListObjectRevisions(ctx context.Context, in *ListObjectRevisionsRequest, opts ...grpc.CallOption) (*ListObjectRevisionsResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListObjectRevisionsResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ListObjectRevisions_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) GetObjectRange(ctx context.Context, in *GetObjectRangeRequest, opts ...grpc.CallOption) (*GetObjectRangeResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetObjectRangeResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_GetObjectRange_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) RebalanceStreams(ctx context.Context, in *RebalanceStreamsRequest, opts ...grpc.CallOption) (*RebalanceStreamsResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(RebalanceStreamsResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_RebalanceStreams_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateConsumerState(ctx context.Context, in *ReplicateConsumerStateRequest, opts ...grpc.CallOption) (*ReplicateConsumerStateResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateConsumerStateResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateConsumerState_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateSourceTailState(ctx context.Context, in *ReplicateSourceTailStateRequest, opts ...grpc.CallOption) (*ReplicateSourceTailStateResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateSourceTailStateResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateSourceTailState_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateStreamCreate(ctx context.Context, in *ReplicateStreamCreateRequest, opts ...grpc.CallOption) (*ReplicateStreamCreateResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateStreamCreateResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateStreamCreate_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateMessage(ctx context.Context, in *ReplicateMessageRequest, opts ...grpc.CallOption) (*ReplicateMessageResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateMessageResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateMessage_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateStreamDelete(ctx context.Context, in *ReplicateStreamDeleteRequest, opts ...grpc.CallOption) (*ReplicateStreamDeleteResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateStreamDeleteResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateStreamDelete_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateTruncate(ctx context.Context, in *ReplicateTruncateRequest, opts ...grpc.CallOption) (*ReplicateTruncateResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateTruncateResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateTruncate_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateStreamUpdate(ctx context.Context, in *ReplicateStreamUpdateRequest, opts ...grpc.CallOption) (*ReplicateStreamUpdateResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateStreamUpdateResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateStreamUpdate_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } func (c *waymakerStreamsServiceClient) ReplicateWorkQueueAck(ctx context.Context, in *ReplicateWorkQueueAckRequest, opts ...grpc.CallOption) (*ReplicateWorkQueueAckResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ReplicateWorkQueueAckResponse) err := c.cc.Invoke(ctx, WaymakerStreamsService_ReplicateWorkQueueAck_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } return out, nil } // WaymakerStreamsServiceServer is the server API for WaymakerStreamsService service. // All implementations must embed UnimplementedWaymakerStreamsServiceServer // for forward compatibility. type WaymakerStreamsServiceServer interface { // --- Stream lifecycle --- CreateStream(context.Context, *CreateStreamRequest) (*CreateStreamResponse, error) DeleteStream(context.Context, *DeleteStreamRequest) (*DeleteStreamResponse, error) GetStreamInfo(context.Context, *GetStreamInfoRequest) (*GetStreamInfoResponse, error) ListStreams(context.Context, *ListStreamsRequest) (*ListStreamsResponse, error) // Slice 3 cross-stream sources admin: enumerate every // (sourcing, source) tail running on this node, with current // last_sourced_seq + pulled_total + last_error. Useful for // operators auditing the cluster's source topology without // ListStreams + GetStreamInfo per stream. GetStreamSources(context.Context, *GetStreamSourcesRequest) (*GetStreamSourcesResponse, error) // Update the *mutable* subset of a stream's config — the Limits // retention bounds (max_age_ms / max_msgs / max_bytes), the per- // message size cap, and the strict-limits toggle. Immutable fields // (name, subjects_filter, block_size, retention policy type) are // not touched. Lowering a bound triggers an immediate prune to // bring stats under the new limit; the primary fans the resulting // truncation out via `ReplicateTruncate` so secondaries mirror. // Partial-update semantics: only fields explicitly set in the // request are applied; unset fields leave the on-disk value // unchanged. UpdateStream(context.Context, *UpdateStreamRequest) (*UpdateStreamResponse, error) // --- Messages --- Publish(context.Context, *PublishRequest) (*PublishResponse, error) Fetch(context.Context, *FetchRequest) (*FetchResponse, error) Ack(context.Context, *AckRequest) (*AckResponse, error) // Negative-acknowledge: server resets the pending entry's // delivered_at_ms so the next fetch redelivers. `delay_ms` defers // eligibility by that wall-clock window (0 = immediate). The // message's `deliver_count` keeps climbing toward `max_deliver`. Nak(context.Context, *NakRequest) (*NakResponse, error) // Terminal-acknowledge: drop the pending entry permanently // without redelivery, regardless of `max_deliver`. Does NOT // trigger WorkQueue delete — other consumers can still observe // the message. Term(context.Context, *TermRequest) (*TermResponse, error) // Heartbeat-acknowledge: bump delivered_at_ms = now to extend // the ack_wait window. `deliver_count` is unchanged. InProgress(context.Context, *InProgressRequest) (*InProgressResponse, error) // Push-mode delivery: the server fetches in a loop and streams // each delivered message back to the client as it arrives. The // client acks via the unary Ack RPC just like pull-mode. The // stream stays open until the client disconnects, the server // returns an error, or the consumer is deleted. Wakes // immediately on new appends via the storage layer's subscribe // primitive — no polling for empty streams. Subscribe(*SubscribeRequest, grpc.ServerStreamingServer[SubscribeEvent]) error // --- Consumers --- CreateConsumer(context.Context, *CreateConsumerRequest) (*CreateConsumerResponse, error) DeleteConsumer(context.Context, *DeleteConsumerRequest) (*DeleteConsumerResponse, error) ListConsumers(context.Context, *ListConsumersRequest) (*ListConsumersResponse, error) GetConsumerInfo(context.Context, *GetConsumerInfoRequest) (*GetConsumerInfoResponse, error) // --- Rebalancing (Phase 1: operator-driven only) --- // // The current owner of a stream serves its raw redb bytes to a peer // that's pulling the stream over. The handler atomically removes the // stream from its local registry first, refusing the call if any // outside reference is still live (operator must drain writers). On // RPC success the source deletes the local file. See // STREAMS_SPEC.md §11 for the model and limitations (no automatic // ring-change sweep yet; the operator is responsible for triggering // a migrate when membership moves a stream's authority). TransferStream(*TransferStreamRequest, grpc.ServerStreamingServer[TransferStreamChunk]) error // Admin trigger on the receiving side: pull stream `name` from // `source_node_id`'s `TransferStream` and own it locally. MigrateStream(context.Context, *MigrateStreamRequest) (*MigrateStreamResponse, error) // Cluster-wide stream inventory + skew report. The receiving node // queries every cluster member's local `StreamsRegistry` (via the // existing proxy channel pool) and aggregates the result. Used by // operators to identify hash-skew imbalance before triggering // `RebalanceStreams`. Also exposed via the `wmkr-status` CLI. GetClusterStreamStats(context.Context, *GetClusterStreamStatsRequest) (*GetClusterStreamStatsResponse, error) // Server-streamed admin watch — emits a WatchEvent each time the // local node's state mutates (stream / consumer create / delete / // update). Useful for live dashboards or service-discovery // clients that want to react to topology changes without // polling. Local-only for now: each watcher sees events generated // on the node it connected to. Cluster-wide watch can be built // on top via a fan-out client; the server doesn't fan out // automatically because the events would arrive out of any // single-source ordering anyway under proxy hops. WatchStreams(*WatchStreamsRequest, grpc.ServerStreamingServer[WatchEvent]) error // Read the latest message at a given subject within a stream. // The foundation for KV-style "last-value wins" lookups on top // of a stream — KV put = Publish to `.`; KV get = // this RPC against the same subject. Returns the full // MessagePb (including headers) so callers can detect KV // tombstones (`wmkv.tombstone` header). // // Returns `success: true` with `message` unset when no message // has ever been published at this subject (or all have been // pruned). Routes via `try_route!` like every other per-stream // RPC. ReadLatestAtSubject(context.Context, *ReadLatestAtSubjectRequest) (*ReadLatestAtSubjectResponse, error) // List every distinct subject in `stream` whose name starts // with `prefix`. Cost is O(matching subjects); independent of // message count. The foundation for `streams-cli kv-keys` and // service-discovery-style "everything under this namespace" // lookups. Returns subjects whose latest message is a // tombstone too — clients that want live-keys-only filter // tombstones via a follow-up `ReadLatestAtSubject`. ListSubjectsByPrefix(context.Context, *ListSubjectsByPrefixRequest) (*ListSubjectsByPrefixResponse, error) // Scan all messages published at an exact subject within // `stream`, in seq order, starting at `from_seq` (0 = from the // beginning), bounded by `limit`. The foundation for // `streams-cli kv-history` — operators want to inspect every // value ever published under a KV key (including tombstones) // for debugging/audit. Cost is O(matching messages); independent // of total stream size. Routes via `try_route!` like every // other per-stream RPC. ScanExactAtSubject(context.Context, *ScanExactAtSubjectRequest) (*ScanExactAtSubjectResponse, error) // Remove a Phase 3 per-stream authority override. Routing // reverts to the ring's hash owner. Idempotent: clearing a // stream with no override succeeds silently. Operators use this // to retire a stale override (e.g. after a ring shift made the // override redundant). Commits via a Raft entry so the clear // applies on every node before the response returns. ClearStreamAuthority(context.Context, *ClearStreamAuthorityRequest) (*ClearStreamAuthorityResponse, error) // List every Phase 3 stream_authority override active on the // responding node. The map is Raft-replicated, so any node's // response reflects the cluster-wide view (modulo apply lag). // Useful for ops triage when an unexpected number of overrides // shows up on /metrics. No fan-out — single-node RPC; the // returned set is the canonical truth. ListStreamAuthorityOverrides(context.Context, *ListStreamAuthorityOverridesRequest) (*ListStreamAuthorityOverridesResponse, error) // Toggle pinned state for `stream`. Pinned streams are exempt // from the auto-GC sweep that retires redundant overrides — use // when you want a stream to stay on its current authority node // even if the ring shifts to make the override redundant. // Idempotent. Independent of the override itself (pinning a // stream with no override is benign; the marker sits dormant). SetStreamPinned(context.Context, *SetStreamPinnedRequest) (*SetStreamPinnedResponse, error) PutObject(context.Context, *PutObjectRequest) (*PutObjectResponse, error) GetObject(context.Context, *GetObjectRequest) (*GetObjectResponse, error) DeleteObject(context.Context, *DeleteObjectRequest) (*DeleteObjectResponse, error) GetObjectInfo(context.Context, *GetObjectInfoRequest) (*GetObjectInfoResponse, error) ListObjects(context.Context, *ListObjectsRequest) (*ListObjectsResponse, error) // Client-streamed PutObject for arbitrary-size objects. First // frame MUST set `start { bucket, name, chunk_size, headers, // sha256 }`. Subsequent frames carry `data` only — each frame's // `data` is ONE chunk message at `objc..`. The server // accumulates a running SHA-256 and total-byte count, publishes // chunks as they arrive (replication fires async), and on the // last frame (`finish=true`) publishes the metadata. A client // disconnect before `finish=true` leaves orphan chunks; the GC // sweep cleans them up. PutObjectStream(grpc.ClientStreamingServer[PutObjectStreamFrame, PutObjectResponse]) error // Server-streamed GetObject. First frame carries `info`; // subsequent frames carry `data` only — one per chunk. Last // frame sets `done=true`. The client reassembles; the response // is sent over the wire in chunk-sized pieces so memory usage // stays bounded on both sides. GetObjectStream(*GetObjectRequest, grpc.ServerStreamingServer[GetObjectStreamFrame]) error // Every revision of `name`'s metadata in seq order — covers // overwrites + tombstones. Returns one entry per metadata // message at `objm.`. Chunks are not enumerated; this RPC // is for object versioning / audit, not for binary diffing. ListObjectRevisions(context.Context, *ListObjectRevisionsRequest) (*ListObjectRevisionsResponse, error) // Read a byte range `[offset, offset + len)` from an object's // assembled payload. Only the chunks that intersect the range // are loaded server-side — useful for resumable downloads of // large objects. // - `offset + len > total_bytes` → returns whatever bytes exist // in the range (success, possibly empty). // - `offset > total_bytes` → returns empty payload (success). // - `len == 0` → returns empty payload (success). GetObjectRange(context.Context, *GetObjectRangeRequest) (*GetObjectRangeResponse, error) // Operator-driven rebalance. Takes an explicit plan — a list of // (stream, target_node) — and executes each step by issuing a // `MigrateStream` to the target. The plan is *not* auto-generated; // the operator (or a future automatic planner) is responsible for // building it from a `GetClusterStreamStats` snapshot. Steps run // sequentially with a per-step timeout; the response carries // per-step outcomes so partial success is visible. RebalanceStreams(context.Context, *RebalanceStreamsRequest) (*RebalanceStreamsResponse, error) // --- Consumer-state replication (Phase 2 §G) --- // // The primary for a stream pushes its consumers' full state to the // stream's `replication_factor - 1` secondaries after every // state-mutating consumer operation (create_consumer, fetch, ack, // delete_consumer). The push is fire-and-forget on the primary's // side — the client RPC has already returned to the caller; the // replication runs in a background task. Secondaries hold the // snapshot in memory; adoption-on-failover is a future slice. ReplicateConsumerState(context.Context, *ReplicateConsumerStateRequest) (*ReplicateConsumerStateResponse, error) // --- Cross-stream sources state replication (slice 2E) --- // // The primary for a sourcing stream pushes the current per-source // tail watermark to each secondary after every successful batch // (i.e. once per ~128 source messages). Secondaries persist the // snapshot via their own SourceTailStore so that on adoption (ring // shift → secondary becomes primary), `spawn_source_tail_tasks` // reads the replicated state and resumes from `last_sourced_seq + 1` // instead of re-pulling from `start_seq` (which would emit // duplicates with already-replicated provenance headers). ReplicateSourceTailState(context.Context, *ReplicateSourceTailStateRequest) (*ReplicateSourceTailStateResponse, error) // --- Stream-data replication (Phase 3, chunk 1) --- // // The primary for a stream pushes: // 1. ReplicateStreamCreate once at create time, so secondaries // know what stream to open in their replica registry with // what config (block_size, retention, max_msg_bytes, etc.). // 2. ReplicateMessage on every successful Publish, with the // seq the primary assigned, so the secondary's replica // mirrors the message log by seq exactly. // // Replica streams live in a per-node "replica registry" rooted at // `/replicas/.redb`, distinct from the // primary-owned namespace. The streams handler never serves // client requests from the replica — it's purely catastrophe // recovery state until the (future) adoption-on-failover slice // promotes a replica to primary. ReplicateStreamCreate(context.Context, *ReplicateStreamCreateRequest) (*ReplicateStreamCreateResponse, error) ReplicateMessage(context.Context, *ReplicateMessageRequest) (*ReplicateMessageResponse, error) // Tear down the replica when the primary deletes the stream. // Idempotent — missing replica is success. ReplicateStreamDelete(context.Context, *ReplicateStreamDeleteRequest) (*ReplicateStreamDeleteResponse, error) // The primary's retention sweep removed messages below // `first_seq`; the secondary mirrors the same truncation so its // replica's first_seq advances in lockstep. Idempotent. ReplicateTruncate(context.Context, *ReplicateTruncateRequest) (*ReplicateTruncateResponse, error) // The primary applied an UpdateStream; secondaries mirror the // mutable subset of the config so a future failover lands on a // replica whose retention matches the primary's. Carries the same // narrow shape as UpdateStreamRequest — only the mutable fields, // with partial-update semantics. ReplicateStreamUpdate(context.Context, *ReplicateStreamUpdateRequest) (*ReplicateStreamUpdateResponse, error) // Under `RetentionPolicy::WorkQueue` the primary deletes a message // on ack (delete-on-first-ack). Without this fan-out, secondaries' // replica files would still hold the acked message — and after a // failover, a fresh consumer on the new primary would see it and // re-deliver, breaking the "each message belongs to exactly one // consumer at a time" invariant. Idempotent: missing seq on // secondary is success. ReplicateWorkQueueAck(context.Context, *ReplicateWorkQueueAckRequest) (*ReplicateWorkQueueAckResponse, error) mustEmbedUnimplementedWaymakerStreamsServiceServer() } // UnimplementedWaymakerStreamsServiceServer must be embedded to have // forward compatible implementations. // // NOTE: this should be embedded by value instead of pointer to avoid a nil // pointer dereference when methods are called. type UnimplementedWaymakerStreamsServiceServer struct{} func (UnimplementedWaymakerStreamsServiceServer) CreateStream(context.Context, *CreateStreamRequest) (*CreateStreamResponse, error) { return nil, status.Error(codes.Unimplemented, "method CreateStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) DeleteStream(context.Context, *DeleteStreamRequest) (*DeleteStreamResponse, error) { return nil, status.Error(codes.Unimplemented, "method DeleteStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetStreamInfo(context.Context, *GetStreamInfoRequest) (*GetStreamInfoResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetStreamInfo not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ListStreams(context.Context, *ListStreamsRequest) (*ListStreamsResponse, error) { return nil, status.Error(codes.Unimplemented, "method ListStreams not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetStreamSources(context.Context, *GetStreamSourcesRequest) (*GetStreamSourcesResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetStreamSources not implemented") } func (UnimplementedWaymakerStreamsServiceServer) UpdateStream(context.Context, *UpdateStreamRequest) (*UpdateStreamResponse, error) { return nil, status.Error(codes.Unimplemented, "method UpdateStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) Publish(context.Context, *PublishRequest) (*PublishResponse, error) { return nil, status.Error(codes.Unimplemented, "method Publish not implemented") } func (UnimplementedWaymakerStreamsServiceServer) Fetch(context.Context, *FetchRequest) (*FetchResponse, error) { return nil, status.Error(codes.Unimplemented, "method Fetch not implemented") } func (UnimplementedWaymakerStreamsServiceServer) Ack(context.Context, *AckRequest) (*AckResponse, error) { return nil, status.Error(codes.Unimplemented, "method Ack not implemented") } func (UnimplementedWaymakerStreamsServiceServer) Nak(context.Context, *NakRequest) (*NakResponse, error) { return nil, status.Error(codes.Unimplemented, "method Nak not implemented") } func (UnimplementedWaymakerStreamsServiceServer) Term(context.Context, *TermRequest) (*TermResponse, error) { return nil, status.Error(codes.Unimplemented, "method Term not implemented") } func (UnimplementedWaymakerStreamsServiceServer) InProgress(context.Context, *InProgressRequest) (*InProgressResponse, error) { return nil, status.Error(codes.Unimplemented, "method InProgress not implemented") } func (UnimplementedWaymakerStreamsServiceServer) Subscribe(*SubscribeRequest, grpc.ServerStreamingServer[SubscribeEvent]) error { return status.Error(codes.Unimplemented, "method Subscribe not implemented") } func (UnimplementedWaymakerStreamsServiceServer) CreateConsumer(context.Context, *CreateConsumerRequest) (*CreateConsumerResponse, error) { return nil, status.Error(codes.Unimplemented, "method CreateConsumer not implemented") } func (UnimplementedWaymakerStreamsServiceServer) DeleteConsumer(context.Context, *DeleteConsumerRequest) (*DeleteConsumerResponse, error) { return nil, status.Error(codes.Unimplemented, "method DeleteConsumer not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ListConsumers(context.Context, *ListConsumersRequest) (*ListConsumersResponse, error) { return nil, status.Error(codes.Unimplemented, "method ListConsumers not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetConsumerInfo(context.Context, *GetConsumerInfoRequest) (*GetConsumerInfoResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetConsumerInfo not implemented") } func (UnimplementedWaymakerStreamsServiceServer) TransferStream(*TransferStreamRequest, grpc.ServerStreamingServer[TransferStreamChunk]) error { return status.Error(codes.Unimplemented, "method TransferStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) MigrateStream(context.Context, *MigrateStreamRequest) (*MigrateStreamResponse, error) { return nil, status.Error(codes.Unimplemented, "method MigrateStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetClusterStreamStats(context.Context, *GetClusterStreamStatsRequest) (*GetClusterStreamStatsResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetClusterStreamStats not implemented") } func (UnimplementedWaymakerStreamsServiceServer) WatchStreams(*WatchStreamsRequest, grpc.ServerStreamingServer[WatchEvent]) error { return status.Error(codes.Unimplemented, "method WatchStreams not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReadLatestAtSubject(context.Context, *ReadLatestAtSubjectRequest) (*ReadLatestAtSubjectResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReadLatestAtSubject not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ListSubjectsByPrefix(context.Context, *ListSubjectsByPrefixRequest) (*ListSubjectsByPrefixResponse, error) { return nil, status.Error(codes.Unimplemented, "method ListSubjectsByPrefix not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ScanExactAtSubject(context.Context, *ScanExactAtSubjectRequest) (*ScanExactAtSubjectResponse, error) { return nil, status.Error(codes.Unimplemented, "method ScanExactAtSubject not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ClearStreamAuthority(context.Context, *ClearStreamAuthorityRequest) (*ClearStreamAuthorityResponse, error) { return nil, status.Error(codes.Unimplemented, "method ClearStreamAuthority not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ListStreamAuthorityOverrides(context.Context, *ListStreamAuthorityOverridesRequest) (*ListStreamAuthorityOverridesResponse, error) { return nil, status.Error(codes.Unimplemented, "method ListStreamAuthorityOverrides not implemented") } func (UnimplementedWaymakerStreamsServiceServer) SetStreamPinned(context.Context, *SetStreamPinnedRequest) (*SetStreamPinnedResponse, error) { return nil, status.Error(codes.Unimplemented, "method SetStreamPinned not implemented") } func (UnimplementedWaymakerStreamsServiceServer) PutObject(context.Context, *PutObjectRequest) (*PutObjectResponse, error) { return nil, status.Error(codes.Unimplemented, "method PutObject not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetObject(context.Context, *GetObjectRequest) (*GetObjectResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetObject not implemented") } func (UnimplementedWaymakerStreamsServiceServer) DeleteObject(context.Context, *DeleteObjectRequest) (*DeleteObjectResponse, error) { return nil, status.Error(codes.Unimplemented, "method DeleteObject not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetObjectInfo(context.Context, *GetObjectInfoRequest) (*GetObjectInfoResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetObjectInfo not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ListObjects(context.Context, *ListObjectsRequest) (*ListObjectsResponse, error) { return nil, status.Error(codes.Unimplemented, "method ListObjects not implemented") } func (UnimplementedWaymakerStreamsServiceServer) PutObjectStream(grpc.ClientStreamingServer[PutObjectStreamFrame, PutObjectResponse]) error { return status.Error(codes.Unimplemented, "method PutObjectStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetObjectStream(*GetObjectRequest, grpc.ServerStreamingServer[GetObjectStreamFrame]) error { return status.Error(codes.Unimplemented, "method GetObjectStream not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ListObjectRevisions(context.Context, *ListObjectRevisionsRequest) (*ListObjectRevisionsResponse, error) { return nil, status.Error(codes.Unimplemented, "method ListObjectRevisions not implemented") } func (UnimplementedWaymakerStreamsServiceServer) GetObjectRange(context.Context, *GetObjectRangeRequest) (*GetObjectRangeResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetObjectRange not implemented") } func (UnimplementedWaymakerStreamsServiceServer) RebalanceStreams(context.Context, *RebalanceStreamsRequest) (*RebalanceStreamsResponse, error) { return nil, status.Error(codes.Unimplemented, "method RebalanceStreams not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateConsumerState(context.Context, *ReplicateConsumerStateRequest) (*ReplicateConsumerStateResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateConsumerState not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateSourceTailState(context.Context, *ReplicateSourceTailStateRequest) (*ReplicateSourceTailStateResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateSourceTailState not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateStreamCreate(context.Context, *ReplicateStreamCreateRequest) (*ReplicateStreamCreateResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateStreamCreate not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateMessage(context.Context, *ReplicateMessageRequest) (*ReplicateMessageResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateMessage not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateStreamDelete(context.Context, *ReplicateStreamDeleteRequest) (*ReplicateStreamDeleteResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateStreamDelete not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateTruncate(context.Context, *ReplicateTruncateRequest) (*ReplicateTruncateResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateTruncate not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateStreamUpdate(context.Context, *ReplicateStreamUpdateRequest) (*ReplicateStreamUpdateResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateStreamUpdate not implemented") } func (UnimplementedWaymakerStreamsServiceServer) ReplicateWorkQueueAck(context.Context, *ReplicateWorkQueueAckRequest) (*ReplicateWorkQueueAckResponse, error) { return nil, status.Error(codes.Unimplemented, "method ReplicateWorkQueueAck not implemented") } func (UnimplementedWaymakerStreamsServiceServer) mustEmbedUnimplementedWaymakerStreamsServiceServer() { } func (UnimplementedWaymakerStreamsServiceServer) testEmbeddedByValue() {} // UnsafeWaymakerStreamsServiceServer may be embedded to opt out of forward compatibility for this service. // Use of this interface is not recommended, as added methods to WaymakerStreamsServiceServer will // result in compilation errors. type UnsafeWaymakerStreamsServiceServer interface { mustEmbedUnimplementedWaymakerStreamsServiceServer() } func RegisterWaymakerStreamsServiceServer(s grpc.ServiceRegistrar, srv WaymakerStreamsServiceServer) { // If the following call panics, it indicates UnimplementedWaymakerStreamsServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { t.testEmbeddedByValue() } s.RegisterService(&WaymakerStreamsService_ServiceDesc, srv) } func _WaymakerStreamsService_CreateStream_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(CreateStreamRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).CreateStream(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_CreateStream_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).CreateStream(ctx, req.(*CreateStreamRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_DeleteStream_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(DeleteStreamRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).DeleteStream(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_DeleteStream_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).DeleteStream(ctx, req.(*DeleteStreamRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetStreamInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetStreamInfoRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetStreamInfo(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetStreamInfo_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetStreamInfo(ctx, req.(*GetStreamInfoRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_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.(WaymakerStreamsServiceServer).ListStreams(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ListStreams_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ListStreams(ctx, req.(*ListStreamsRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetStreamSources_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetStreamSourcesRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetStreamSources(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetStreamSources_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetStreamSources(ctx, req.(*GetStreamSourcesRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_UpdateStream_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(UpdateStreamRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).UpdateStream(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_UpdateStream_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).UpdateStream(ctx, req.(*UpdateStreamRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_Publish_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(PublishRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).Publish(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_Publish_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).Publish(ctx, req.(*PublishRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_Fetch_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(FetchRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).Fetch(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_Fetch_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).Fetch(ctx, req.(*FetchRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_Ack_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(AckRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).Ack(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_Ack_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).Ack(ctx, req.(*AckRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_Nak_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(NakRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).Nak(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_Nak_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).Nak(ctx, req.(*NakRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_Term_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(TermRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).Term(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_Term_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).Term(ctx, req.(*TermRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_InProgress_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(InProgressRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).InProgress(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_InProgress_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).InProgress(ctx, req.(*InProgressRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(SubscribeRequest) if err := stream.RecvMsg(m); err != nil { return err } return srv.(WaymakerStreamsServiceServer).Subscribe(m, &grpc.GenericServerStream[SubscribeRequest, SubscribeEvent]{ServerStream: stream}) } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type WaymakerStreamsService_SubscribeServer = grpc.ServerStreamingServer[SubscribeEvent] func _WaymakerStreamsService_CreateConsumer_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(CreateConsumerRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).CreateConsumer(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_CreateConsumer_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).CreateConsumer(ctx, req.(*CreateConsumerRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_DeleteConsumer_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(DeleteConsumerRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).DeleteConsumer(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_DeleteConsumer_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).DeleteConsumer(ctx, req.(*DeleteConsumerRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ListConsumers_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ListConsumersRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ListConsumers(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ListConsumers_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ListConsumers(ctx, req.(*ListConsumersRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetConsumerInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetConsumerInfoRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetConsumerInfo(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetConsumerInfo_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetConsumerInfo(ctx, req.(*GetConsumerInfoRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_TransferStream_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(TransferStreamRequest) if err := stream.RecvMsg(m); err != nil { return err } return srv.(WaymakerStreamsServiceServer).TransferStream(m, &grpc.GenericServerStream[TransferStreamRequest, TransferStreamChunk]{ServerStream: stream}) } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type WaymakerStreamsService_TransferStreamServer = grpc.ServerStreamingServer[TransferStreamChunk] func _WaymakerStreamsService_MigrateStream_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(MigrateStreamRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).MigrateStream(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_MigrateStream_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).MigrateStream(ctx, req.(*MigrateStreamRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetClusterStreamStats_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetClusterStreamStatsRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetClusterStreamStats(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetClusterStreamStats_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetClusterStreamStats(ctx, req.(*GetClusterStreamStatsRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_WatchStreams_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(WatchStreamsRequest) if err := stream.RecvMsg(m); err != nil { return err } return srv.(WaymakerStreamsServiceServer).WatchStreams(m, &grpc.GenericServerStream[WatchStreamsRequest, WatchEvent]{ServerStream: stream}) } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type WaymakerStreamsService_WatchStreamsServer = grpc.ServerStreamingServer[WatchEvent] func _WaymakerStreamsService_ReadLatestAtSubject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReadLatestAtSubjectRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReadLatestAtSubject(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReadLatestAtSubject_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReadLatestAtSubject(ctx, req.(*ReadLatestAtSubjectRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ListSubjectsByPrefix_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ListSubjectsByPrefixRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ListSubjectsByPrefix(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ListSubjectsByPrefix_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ListSubjectsByPrefix(ctx, req.(*ListSubjectsByPrefixRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ScanExactAtSubject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ScanExactAtSubjectRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ScanExactAtSubject(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ScanExactAtSubject_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ScanExactAtSubject(ctx, req.(*ScanExactAtSubjectRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ClearStreamAuthority_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ClearStreamAuthorityRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ClearStreamAuthority(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ClearStreamAuthority_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ClearStreamAuthority(ctx, req.(*ClearStreamAuthorityRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ListStreamAuthorityOverrides_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ListStreamAuthorityOverridesRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ListStreamAuthorityOverrides(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ListStreamAuthorityOverrides_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ListStreamAuthorityOverrides(ctx, req.(*ListStreamAuthorityOverridesRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_SetStreamPinned_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(SetStreamPinnedRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).SetStreamPinned(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_SetStreamPinned_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).SetStreamPinned(ctx, req.(*SetStreamPinnedRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_PutObject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(PutObjectRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).PutObject(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_PutObject_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).PutObject(ctx, req.(*PutObjectRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetObject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetObjectRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetObject(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetObject_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetObject(ctx, req.(*GetObjectRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_DeleteObject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(DeleteObjectRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).DeleteObject(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_DeleteObject_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).DeleteObject(ctx, req.(*DeleteObjectRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetObjectInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetObjectInfoRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetObjectInfo(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetObjectInfo_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetObjectInfo(ctx, req.(*GetObjectInfoRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ListObjects_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ListObjectsRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ListObjects(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ListObjects_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ListObjects(ctx, req.(*ListObjectsRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_PutObjectStream_Handler(srv interface{}, stream grpc.ServerStream) error { return srv.(WaymakerStreamsServiceServer).PutObjectStream(&grpc.GenericServerStream[PutObjectStreamFrame, PutObjectResponse]{ServerStream: stream}) } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type WaymakerStreamsService_PutObjectStreamServer = grpc.ClientStreamingServer[PutObjectStreamFrame, PutObjectResponse] func _WaymakerStreamsService_GetObjectStream_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(GetObjectRequest) if err := stream.RecvMsg(m); err != nil { return err } return srv.(WaymakerStreamsServiceServer).GetObjectStream(m, &grpc.GenericServerStream[GetObjectRequest, GetObjectStreamFrame]{ServerStream: stream}) } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type WaymakerStreamsService_GetObjectStreamServer = grpc.ServerStreamingServer[GetObjectStreamFrame] func _WaymakerStreamsService_ListObjectRevisions_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ListObjectRevisionsRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ListObjectRevisions(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ListObjectRevisions_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ListObjectRevisions(ctx, req.(*ListObjectRevisionsRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_GetObjectRange_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetObjectRangeRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).GetObjectRange(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_GetObjectRange_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).GetObjectRange(ctx, req.(*GetObjectRangeRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_RebalanceStreams_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(RebalanceStreamsRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).RebalanceStreams(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_RebalanceStreams_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).RebalanceStreams(ctx, req.(*RebalanceStreamsRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateConsumerState_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateConsumerStateRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateConsumerState(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateConsumerState_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateConsumerState(ctx, req.(*ReplicateConsumerStateRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateSourceTailState_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateSourceTailStateRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateSourceTailState(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateSourceTailState_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateSourceTailState(ctx, req.(*ReplicateSourceTailStateRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateStreamCreate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateStreamCreateRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateStreamCreate(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateStreamCreate_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateStreamCreate(ctx, req.(*ReplicateStreamCreateRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateMessage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateMessageRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateMessage(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateMessage_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateMessage(ctx, req.(*ReplicateMessageRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateStreamDelete_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateStreamDeleteRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateStreamDelete(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateStreamDelete_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateStreamDelete(ctx, req.(*ReplicateStreamDeleteRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateTruncate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateTruncateRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateTruncate(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateTruncate_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateTruncate(ctx, req.(*ReplicateTruncateRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateStreamUpdate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateStreamUpdateRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateStreamUpdate(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateStreamUpdate_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateStreamUpdate(ctx, req.(*ReplicateStreamUpdateRequest)) } return interceptor(ctx, in, info, handler) } func _WaymakerStreamsService_ReplicateWorkQueueAck_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ReplicateWorkQueueAckRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(WaymakerStreamsServiceServer).ReplicateWorkQueueAck(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: WaymakerStreamsService_ReplicateWorkQueueAck_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(WaymakerStreamsServiceServer).ReplicateWorkQueueAck(ctx, req.(*ReplicateWorkQueueAckRequest)) } return interceptor(ctx, in, info, handler) } // WaymakerStreamsService_ServiceDesc is the grpc.ServiceDesc for WaymakerStreamsService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) var WaymakerStreamsService_ServiceDesc = grpc.ServiceDesc{ ServiceName: "waymaker.streams.WaymakerStreamsService", HandlerType: (*WaymakerStreamsServiceServer)(nil), Methods: []grpc.MethodDesc{ { MethodName: "CreateStream", Handler: _WaymakerStreamsService_CreateStream_Handler, }, { MethodName: "DeleteStream", Handler: _WaymakerStreamsService_DeleteStream_Handler, }, { MethodName: "GetStreamInfo", Handler: _WaymakerStreamsService_GetStreamInfo_Handler, }, { MethodName: "ListStreams", Handler: _WaymakerStreamsService_ListStreams_Handler, }, { MethodName: "GetStreamSources", Handler: _WaymakerStreamsService_GetStreamSources_Handler, }, { MethodName: "UpdateStream", Handler: _WaymakerStreamsService_UpdateStream_Handler, }, { MethodName: "Publish", Handler: _WaymakerStreamsService_Publish_Handler, }, { MethodName: "Fetch", Handler: _WaymakerStreamsService_Fetch_Handler, }, { MethodName: "Ack", Handler: _WaymakerStreamsService_Ack_Handler, }, { MethodName: "Nak", Handler: _WaymakerStreamsService_Nak_Handler, }, { MethodName: "Term", Handler: _WaymakerStreamsService_Term_Handler, }, { MethodName: "InProgress", Handler: _WaymakerStreamsService_InProgress_Handler, }, { MethodName: "CreateConsumer", Handler: _WaymakerStreamsService_CreateConsumer_Handler, }, { MethodName: "DeleteConsumer", Handler: _WaymakerStreamsService_DeleteConsumer_Handler, }, { MethodName: "ListConsumers", Handler: _WaymakerStreamsService_ListConsumers_Handler, }, { MethodName: "GetConsumerInfo", Handler: _WaymakerStreamsService_GetConsumerInfo_Handler, }, { MethodName: "MigrateStream", Handler: _WaymakerStreamsService_MigrateStream_Handler, }, { MethodName: "GetClusterStreamStats", Handler: _WaymakerStreamsService_GetClusterStreamStats_Handler, }, { MethodName: "ReadLatestAtSubject", Handler: _WaymakerStreamsService_ReadLatestAtSubject_Handler, }, { MethodName: "ListSubjectsByPrefix", Handler: _WaymakerStreamsService_ListSubjectsByPrefix_Handler, }, { MethodName: "ScanExactAtSubject", Handler: _WaymakerStreamsService_ScanExactAtSubject_Handler, }, { MethodName: "ClearStreamAuthority", Handler: _WaymakerStreamsService_ClearStreamAuthority_Handler, }, { MethodName: "ListStreamAuthorityOverrides", Handler: _WaymakerStreamsService_ListStreamAuthorityOverrides_Handler, }, { MethodName: "SetStreamPinned", Handler: _WaymakerStreamsService_SetStreamPinned_Handler, }, { MethodName: "PutObject", Handler: _WaymakerStreamsService_PutObject_Handler, }, { MethodName: "GetObject", Handler: _WaymakerStreamsService_GetObject_Handler, }, { MethodName: "DeleteObject", Handler: _WaymakerStreamsService_DeleteObject_Handler, }, { MethodName: "GetObjectInfo", Handler: _WaymakerStreamsService_GetObjectInfo_Handler, }, { MethodName: "ListObjects", Handler: _WaymakerStreamsService_ListObjects_Handler, }, { MethodName: "ListObjectRevisions", Handler: _WaymakerStreamsService_ListObjectRevisions_Handler, }, { MethodName: "GetObjectRange", Handler: _WaymakerStreamsService_GetObjectRange_Handler, }, { MethodName: "RebalanceStreams", Handler: _WaymakerStreamsService_RebalanceStreams_Handler, }, { MethodName: "ReplicateConsumerState", Handler: _WaymakerStreamsService_ReplicateConsumerState_Handler, }, { MethodName: "ReplicateSourceTailState", Handler: _WaymakerStreamsService_ReplicateSourceTailState_Handler, }, { MethodName: "ReplicateStreamCreate", Handler: _WaymakerStreamsService_ReplicateStreamCreate_Handler, }, { MethodName: "ReplicateMessage", Handler: _WaymakerStreamsService_ReplicateMessage_Handler, }, { MethodName: "ReplicateStreamDelete", Handler: _WaymakerStreamsService_ReplicateStreamDelete_Handler, }, { MethodName: "ReplicateTruncate", Handler: _WaymakerStreamsService_ReplicateTruncate_Handler, }, { MethodName: "ReplicateStreamUpdate", Handler: _WaymakerStreamsService_ReplicateStreamUpdate_Handler, }, { MethodName: "ReplicateWorkQueueAck", Handler: _WaymakerStreamsService_ReplicateWorkQueueAck_Handler, }, }, Streams: []grpc.StreamDesc{ { StreamName: "Subscribe", Handler: _WaymakerStreamsService_Subscribe_Handler, ServerStreams: true, }, { StreamName: "TransferStream", Handler: _WaymakerStreamsService_TransferStream_Handler, ServerStreams: true, }, { StreamName: "WatchStreams", Handler: _WaymakerStreamsService_WatchStreams_Handler, ServerStreams: true, }, { StreamName: "PutObjectStream", Handler: _WaymakerStreamsService_PutObjectStream_Handler, ClientStreams: true, }, { StreamName: "GetObjectStream", Handler: _WaymakerStreamsService_GetObjectStream_Handler, ServerStreams: true, }, }, Metadata: "waymaker_streams.proto", }