# Streams Client Accessed via `client.Streams()` on the [Hatchet client](/reference/go/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. ```go func (s *StreamsClient) Publish(ctx context.Context, topic string, message []byte, opts ...StreamCallOpt) error ``` Parameters: 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. ```go func (s *StreamsClient) Reset(ctx context.Context, topic string, opts ...StreamCallOpt) error ``` Parameters: 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. ```go 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. ```go 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. ```go func WithStreamBatchTimeout(d time.Duration) StreamCallOpt ``` Parameters: 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. ```go func WithStreamCursor(cursor string) StreamCallOpt ``` Parameters: Name, Type `cursor`, `string` Returns: Type `StreamCallOpt` ##### `WithStreamNamespace` WithStreamNamespace sets the topic's namespace; the empty namespace when omitted. ```go func WithStreamNamespace(namespace string) StreamCallOpt ``` Parameters: 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` ```go func (e *StreamPayloadError) Error() string ``` Returns: Type `string` ##### `Unwrap` ```go func (e *StreamPayloadError) Unwrap() error ``` Returns: Type `error`