Streams ClientGo SDK
Accessed via client.Streams() on the Hatchet client.
StreamsClient
StreamsClient publishes to and reads from durable topics. Unlike Runs().SubscribeToStream, topics are stored, independent of any run, and readers can resume from a cursor.
Methods:
| Name | Description |
|---|---|
Publish | Publish returns once the message is stored. |
Reset | Reset makes Subscribe calls without a cursor start after WithStreamCursor's message, or the topic's latest message. |
Subscribe | Subscribe yields topic's messages after WithStreamCursor's message, or from the oldest retained (or, once the topic is reset, after the reset). |
TopicMetadata | TopicMetadata describes topic's retained messages. |
Functions
Publish
Publish returns once the message is stored. A client's messages to a topic are delivered in the order Publish was called. Publishes to a topic share requests: everything queued while one is in flight goes out in the next, and WithStreamBatchTimeout holds a message up to that long for more to join it. If ctx ends after the message was sent, Publish returns ctx's error though the message may still be stored.
func (s *StreamsClient) Publish(ctx context.Context, topic string, message []byte, opts ...StreamCallOpt) errorParameters:
| Name | Type |
|---|---|
ctx | context.Context |
topic | string |
message | []byte |
opts | ...StreamCallOpt |
Returns:
| Type |
|---|
error |
Reset
Reset makes Subscribe calls without a cursor start after WithStreamCursor's message, or the topic's latest message. Readers already iterating, and any call with a cursor, are unaffected.
func (s *StreamsClient) Reset(ctx context.Context, topic string, opts ...StreamCallOpt) errorParameters:
| Name | Type |
|---|---|
ctx | context.Context |
topic | string |
opts | ...StreamCallOpt |
Returns:
| Type |
|---|
error |
Subscribe
Subscribe yields topic's messages after WithStreamCursor's message, or from the oldest retained (or, once the topic is reset, after the reset). If the connection drops, it resubscribes from the last message yielded, so nothing is skipped or repeated. It ends when the server hangs up an idle subscription, ctx ends, or the caller stops iterating; it yields an error, then ends, on errors a retry can't fix, such as an expired cursor.
func (s *StreamsClient) Subscribe(ctx context.Context, topic string, opts ...StreamCallOpt) iter.Seq2[StreamEvent, error]Parameters:
| Name | Type |
|---|---|
ctx | context.Context |
topic | string |
opts | ...StreamCallOpt |
Returns:
| Type |
|---|
iter.Seq2[StreamEvent, error] |
TopicMetadata
TopicMetadata describes topic's retained messages. It returns a NotFound gRPC error for a topic nothing was published to.
func (s *StreamsClient) TopicMetadata(ctx context.Context, topic string, opts ...StreamCallOpt) (*StreamTopicMetadata, error)Parameters:
| Name | Type |
|---|---|
ctx | context.Context |
topic | string |
opts | ...StreamCallOpt |
Returns:
| Type |
|---|
*StreamTopicMetadata |
error |
Related types
StreamEvent
StreamEvent is a message read from a durable stream topic.
Fields:
| Name | Type | Description |
|---|---|---|
Payload | []byte | Payload is the message payload. |
Cursor | string | Cursor resumes Subscribe right after this event. |
CreatedAt | time.Time | CreatedAt is when the message was stored. |
StreamTopicMetadata
StreamTopicMetadata describes a topic's retained messages; counts and the latest message cover the tenant's retention.
Fields:
| Name | Type | Description |
|---|---|---|
Namespace | string | |
Topic | string | |
TenantId | string | |
MessageCount | int64 | |
LatestCursor | *string | LatestCursor resumes Subscribe after the newest retained message; nil when none is retained. |
LastPublishedAt | *time.Time | LastPublishedAt is when the newest retained message was stored; nil when none is retained. |
StreamCallOpt
StreamCallOpt configures a call on the StreamsClient.
Functions
WithStreamBatchTimeout
WithStreamBatchTimeout lets Publish hold the message up to d for later publishes to the same topic to share its request, bounding the latency batching adds. 0, the default, sends it as soon as no other request to the topic is in flight.
func WithStreamBatchTimeout(d time.Duration) StreamCallOptParameters:
| Name | Type |
|---|---|
d | time.Duration |
Returns:
| Type |
|---|
StreamCallOpt |
WithStreamCursor
WithStreamCursor makes Subscribe resume right after the message the cursor came from, or Reset start new readers after it. Ignored by Publish and TopicMetadata.
func WithStreamCursor(cursor string) StreamCallOptParameters:
| Name | Type |
|---|---|
cursor | string |
Returns:
| Type |
|---|
StreamCallOpt |
WithStreamNamespace
WithStreamNamespace sets the topic's namespace; the empty namespace when omitted.
func WithStreamNamespace(namespace string) StreamCallOptParameters:
| Name | Type |
|---|---|
namespace | string |
Returns:
| Type |
|---|
StreamCallOpt |
StreamPayloadError
StreamPayloadError is a failed request for a payload too large for a gRPC message.
Fields:
| Name | Type | Description |
|---|---|---|
StatusCode | int | StatusCode is 0 when the request got no response. |
Err | error |
Methods:
| Name | Description |
|---|---|
Error | |
Unwrap |
Functions
Error
func (e *StreamPayloadError) Error() stringReturns:
| Type |
|---|
string |
Unwrap
func (e *StreamPayloadError) Unwrap() errorReturns:
| Type |
|---|
error |
Last updated on October 29, 2026