riotpiaoleandClaude Haiku 4.5
e4017aea9e
- Add Forgejo CI workflow: gofmt checks, module caching, coverage reporting - Add release workflow: auto-tag-triggered release with changelog extraction - Update module paths from rock/ to homelab/ namespace - Enhance test coverage and documentation (PLAN.md, README.md) Co-Authored-By: Claude Haiku 4.5 <[email protected]>
92 lines
2.8 KiB
Go
92 lines
2.8 KiB
Go
package kmsvc
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
kafkamgmtv1 "forgejo.riotpiao.homelab.com/homelab/kmsvc-proto/gen/kafkamgmt/v1"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
)
|
|
|
|
// Client is a Go client for the Kafka Management Service message-plane API.
|
|
// Queue lifecycle (create/delete/configure) is managed via the Queue CRD, not
|
|
// this client — see kafaka_management_service design.md §2a/§2b.
|
|
type Client struct {
|
|
conn *grpc.ClientConn
|
|
stub kafkamgmtv1.QueueServiceClient
|
|
}
|
|
|
|
// Option configures a Client during New.
|
|
type Option func(*options)
|
|
|
|
type options struct {
|
|
tokenSource TokenSource
|
|
tlsConfig credentials.TransportCredentials
|
|
dialTimeout time.Duration
|
|
dialOpts []grpc.DialOption
|
|
}
|
|
|
|
// WithTokenSource attaches a bearer token to every outgoing call via source.
|
|
func WithTokenSource(source TokenSource) Option {
|
|
return func(o *options) { o.tokenSource = source }
|
|
}
|
|
|
|
// WithTransportCredentials sets the gRPC transport credentials (e.g. TLS).
|
|
// If not set, the connection is plaintext (insecure.NewCredentials()) —
|
|
// appropriate for cluster-internal traffic, not for use over an untrusted
|
|
// network.
|
|
func WithTransportCredentials(creds credentials.TransportCredentials) Option {
|
|
return func(o *options) { o.tlsConfig = creds }
|
|
}
|
|
|
|
// WithDialTimeout bounds how long New waits for the initial connection.
|
|
func WithDialTimeout(d time.Duration) Option {
|
|
return func(o *options) { o.dialTimeout = d }
|
|
}
|
|
|
|
// New dials target (host:port) and returns a ready-to-use Client.
|
|
func New(ctx context.Context, target string, opts ...Option) (*Client, error) {
|
|
o := &options{dialTimeout: 10 * time.Second}
|
|
for _, opt := range opts {
|
|
opt(o)
|
|
}
|
|
|
|
creds := o.tlsConfig
|
|
if creds == nil {
|
|
creds = insecure.NewCredentials()
|
|
}
|
|
|
|
dialOpts := []grpc.DialOption{grpc.WithTransportCredentials(creds)}
|
|
if o.tokenSource != nil {
|
|
dialOpts = append(dialOpts,
|
|
grpc.WithUnaryInterceptor(authUnaryInterceptor(o.tokenSource)),
|
|
grpc.WithStreamInterceptor(authStreamInterceptor(o.tokenSource)),
|
|
)
|
|
}
|
|
dialOpts = append(dialOpts, o.dialOpts...)
|
|
|
|
dialCtx, cancel := context.WithTimeout(ctx, o.dialTimeout)
|
|
defer cancel()
|
|
|
|
conn, err := grpc.DialContext(dialCtx, target, dialOpts...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("kmsvc: dial %s: %w", target, err)
|
|
}
|
|
|
|
return &Client{conn: conn, stub: kafkamgmtv1.NewQueueServiceClient(conn)}, nil
|
|
}
|
|
|
|
// newFromConn builds a Client around an existing connection — used by tests
|
|
// to wire up an in-process bufconn connection without a real dial.
|
|
func newFromConn(conn *grpc.ClientConn) *Client {
|
|
return &Client{conn: conn, stub: kafkamgmtv1.NewQueueServiceClient(conn)}
|
|
}
|
|
|
|
// Close releases the underlying gRPC connection.
|
|
func (c *Client) Close() error {
|
|
return c.conn.Close()
|
|
}
|