mirror of
https://github.com/hanzoai/kafka.git
synced 2026-10-30 14:48:25 +00:00
zeekayandHanzo Dev
c3e231795c
This is the Kafka interface to Hanzo PubSub, not a product of its own. "Stream"
described the transport rather than the contract, and left two names for one
thing -- the repo was already redirecting hanzoai/kafka -> hanzoai/stream, so
imports and docs disagreed with the canonical URL.
The layering it belongs to:
hanzoai/tasks workloads and async workflows
|
hanzoai/pubsub the ONE event bus (JetStream-durable NATS)
|
hanzoai/kafka Kafka-wire adaptor (AMQP is the sibling seam)
|
hanzoai/datastore the ONE warehouse
Module path is now github.com/hanzoai/kafka; binary, image labels and docs say
Hanzo Kafka.
The word "stream" survives wherever it names the PROTOCOL rather than the
product -- JetStream stream names, ParseStreamName, the subject patterns in
pubsub/streams.go. Renaming those would have broken the wire contract to make a
brand consistent.
Unit tests pass. test/cluster and test/e2e still fail identically to before this
change: they shell out to an Apache Kafka distribution at ~/kafka_2.13-3.9.0/
that is not installed here, so they are unrelated to the rename.
Co-authored-by: Hanzo Dev <[email protected]>
38 lines
882 B
Go
38 lines
882 B
Go
package pubsub
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
log "github.com/hanzoai/kafka/logging"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// Client wraps Hanzo PubSub connection and Hanzo Kafka context
|
|
type Client struct {
|
|
NC *nats.Conn
|
|
JS nats.JetStreamContext
|
|
}
|
|
|
|
// NewClient connects to Hanzo PubSub and obtains Hanzo Kafka context
|
|
func NewClient(url string, opts ...nats.Option) (*Client, error) {
|
|
nc, err := nats.Connect(url, opts...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to connect to Hanzo PubSub at %s: %w", url, err)
|
|
}
|
|
js, err := nc.JetStream()
|
|
if err != nil {
|
|
nc.Close()
|
|
return nil, fmt.Errorf("failed to get Hanzo Kafka context: %w", err)
|
|
}
|
|
log.Info("Connected to Hanzo PubSub at %s", url)
|
|
return &Client{NC: nc, JS: js}, nil
|
|
}
|
|
|
|
// Close closes the PubSub connection
|
|
func (c *Client) Close() {
|
|
if c.NC != nil {
|
|
c.NC.Close()
|
|
log.Info("Hanzo PubSub connection closed")
|
|
}
|
|
}
|