ensign

package module
v0.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 23, 2023 License: BSD-3-Clause Imports: 26 Imported by: 22

README

Ensign Go SDK

Go Reference Go Report Card CI

Welcome to go-ensign!

This repository contains the Ensign driver, SDK, and helpers for Go. For the main ensign repo, go here. We also have SDKs for Javascript and Python.

The getting started guide and general documentation can be found at https://ensign.rotational.dev. You may also want to reference the GoDoc Package Documentation as well.

Creating an Ensign Client

To add the Go SDK as a dependency, either go get it or import it and run go mod tidy:

$ go get github.com/rotationalio/go-ensign

The Go SDK provides a client that is able to connect to an Ensign system in order to manage topics, publish events, and subscribe to an event stream. At a minimum, you need API credentials, which can be obtained by creating an account at https://rotational.app. Once you've created an account and downloaded your credentials, you can instantiate a new Ensign client and check to make sure you're connected:

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/rotationalio/go-ensign"
)

func main() {
	client, err := ensign.New(ensign.WithCredentials("CLIENT ID", "CLIENT SECRET"))
	if err != nil {
		log.Fatal(err)
	}

	status, err := client.Status(context.Background())
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("%+v\n", status)
}

You can also set the $ENSIGN_CLIENT_ID and $ENSIGN_CLIENT_SECRET environment variables so that you can instantiate the Client without specifying credentials in code.

// Assumes that $ENSIGN_CLIENT_ID and $ENSIGN_CLIENT_SECRET are set
client, err := ensign.New()

Finally, if you downloaded the client.json file from the app; you can load it by specifying the path to the JSON file:

client, err := ensign.New(ensign.WithLoadCredentials("path/to/client.json"))

Topic Management

Every topic that you work with is like a database table or collection of tables -- it is where your event data lives. Naturally, this is generally a starting place to interacting with Ensign. While you can create and manage topics from Beacon -- our Ensign UI -- you can also create and manage topics with the SDK.

Generally speaking, at the beginning of each Ensign program, you'll want to check if a topic exists and create it if it doesn't. This can be done with the following code:

const TopicName = "my-awesome-topic"

var client *ensign.Client

func checkTopic() (err error) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	var exists bool
	if exists, err = client.TopicExists(ctx, TopicName); err != nil {
		return err
	}

	if !exists {
		var topicID string
		if topicID, err = client.CreateTopic(ctx, TopicName); err != nil {
			return err
		}
		log.Printf("created topic %s with ID %s\n", TopicName, topicID)
	}
	return nil
}

You can also list all the topics in your current project as follows:

func printAllTopics() error {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	topics, err := client.ListTopics(ctx)
	if err != nil {
		return err
	}

	for _, topic := range topics {
		fmt.Printf("%s: %s\n", topic.Id, topic.Name)
	}
	return nil
}
Topic Cache

If you're going to be working with a lot of topics, you can use the topics.Cache to simplify topic management. Create the cache as follows:


import (
	"github.com/rotationalio/ensign"
	ensignTopics "github.com/rotationalio/ensign/topics"
)

var (
	client *ensign.Client
	topics *ensignTopics.Cache
)

func connect() (err error) {
	if client, err = ensign.New(); err != nil {
		return err
	}

	topics = ensignTopics.NewCache(client)
	return nil
}

With the topics cache in place you can simplify the "check if topic exists and create if it doesn't" code by using the Ensure() method:

const TopicName = "my-awesome-topic"

func checkTopic() (err error) {
	if _, err = topics.Ensure(TopicName); err != nil {
		return err
	}
	return nil
}

The cache also prevents repeated calls to Ensign, so you can use the Exists and Get method instead of client.TopicExists and client.TopicID.

Events

The Event data structure is how you can create data to send to Ensign and your downstream subscribers, and how you will receive data from Ensign. At it's core, an event is an application-defined datagram that you can use to create totally ordered data flows.

There are two pieces user-specified information that are part of the event, the Metadata: user-defined key/value pairs of strings that can be used for querying and indexing events, and the Data, generic data that you can use that define the event.

To parse and work with events, there are two pieces of information: the Mimetype, e.g. application/json, which helps you determine how to parse the Data and the Type, a user defined schema that can be used to validate or verify the event once parsed. For more on types and mimetypes, see the Ensign documentation.

Publishing Events

To publish an event to a topic, you create an event, and then use the clients, Publish method as follows:

var client *ensign.Client

func publishMessages() {
	for i := 0; i < 100; i++ {
		// Create a simple event
		msg := &ensign.Event{Data: []byte(fmt.Sprintf("event no. %d", i+1))}}

		// Publish the event
		if err := client.Publish("example-topic", msg); err != nil {
			panic(err)
		}

		// Determine if the event was published or not
		if _, err := msg.Nacked(); err != nil {
			panic(err)
		}

		time.Sleep(time.Second)
	}
}

When publishing events you can check if the event was acked (sucessfully published) or nacked (there was an error during publishing) using the Acked() and Nacked() methods of the Event that you created.

Subscribing

To subscribe to events on a topic or topics, you can create an object with a channel to receive events on.

var client *ensign.Client

func subscribeEvents() {
	sub, err := client.Subscribe("example-topic-a", "example-topic-b")
	if err != nil {
		panic(err)
	}

	for event := range sub.C {
		if err := handleEvent(event); err != nil {
			event.Nack(api.Nack_UNPROCESSED)
		} else {
			event.Ack()
		}
	}
}

func handleEvent(event *ensign.Event) error {
	// handle each event as it comes in.
}

It is important to let Ensign know if the event was processed successfully using the Ack and Nack methods on the event -- this will help Ensign determine if it needs to resend the event or not.

Quick API Reference

  • New: create a new Ensign client with credentials from the environment or from a file.
  • Event: the event data structure for publishing and subscribing.
  • client.Publish: publish one or more events to a specified topic.
  • client.Subscribe: create a Subscription with a channel to receive incoming events on.
  • Subscription: the object return from a Subscribe operation, with an events channel, C to listen for events on.
  • client.TopicExists: check if a topic with the specified name exists in the project.
  • client.CreateTopic: create a topic with the specified name in the project.
  • client.ListTopics: list all the topics in the project.
  • client.TopicID: get the topic ID from a topic name.
  • cache.Get: get a topic ID from a topic name.
  • cache.Exists: check if the specified topic exists.
  • cache.Ensure: a helper for "create topic if it doesn't exist".
  • cache.Clear: empty the internal cache to fetch data from ensign again.
  • cache.Length: returns the number of items in the topic cache.

Documentation

Index

Constants

View Source 
const (
	// Specifies the wait period before checking if a gRPC connection has been
	// established while waiting for a ready connection.
	ReconnectTick = 750 * time.Millisecond

	// The default page size for paginated gRPC responses.
	DefaultPageSize = uint32(100)

	// The Go SDK user agent format string.
	UserAgent = "Ensign Go SDK/v%d"
)
View Source 
const (
	// APIKeys you receive when creating a project on https://rotational.app will return
	// a Client ID and Client Secret. These should be specified in the environment with
	// the following environment variables. You cannot connect to Ensign without these
	// APIKeys. Note that APIKeys only allow connections to one project in Ensign. In
	// the future, Ensign will be able to support multiple key credentials to access
	// multiple projects.
	EnvClientID     = "ENSIGN_CLIENT_ID"
	EnvClientSecret = "ENSIGN_CLIENT_SECRET"

	// The following environment variables allow you to connect to another Ensign
	// service. These are primarily used by the Rotational team to connect to develop
	// and staging Ensign environments. However, if you have an Enterprise deployment of
	// Ensign, you may need to specify these alternative connection details.
	EnvEndpoint = "ENSIGN_ENDPOINT"
	EnvInsecure = "ENSIGN_INSECURE"
	EnvAuthURL  = "ENSIGN_AUTH_URL"
	EnvNoAuth   = "ENSIGN_NO_AUTHENTICATION"
)

Environment variables for configuring Ensign. Unless otherwise specified in the options, Ensign will try to configure your client from the environment.

View Source 
const (
	EnsignEndpoint = "ensign.rotational.app:443"
	AuthEndpoint   = "https://auth.rotational.app"
)

Default connection endpoints to the production Ensign cluster.

View Source 
const (
	VersionMajor         = 0
	VersionMinor         = 12
	VersionPatch         = 0
	VersionReleaseLevel  = "beta"
	VersionReleaseNumber = 11
)

Version component constants for the current build.

Variables

View Source 
var (
	ErrMissingEndpoint     = errors.New("invalid options: endpoint is required")
	ErrMissingClientID     = errors.New("invalid options: client ID is required")
	ErrMissingClientSecret = errors.New("invalid options: client secret is required")
	ErrMissingAuthURL      = errors.New("invalid options: auth url is required")
	ErrMissingMock         = errors.New("invalid options: in testing mode a mock grpc server is required")
	ErrTopicNameNotFound   = errors.New("topic name not found in project")
	ErrCannotAck           = errors.New("cannot ack or nack an event not received from subscribe")
	ErrOverwrite           = errors.New("this operation would overwrite existing event data")
	ErrNoTopicID           = errors.New("topic id is not available on event")
	ErrEmptyQuery          = errors.New("query cannot be empty")
	ErrCursorClosed        = errors.New("cursor is closed")
	ErrTopicInfoNotFound   = errors.New("no info found for specified topic")
	ErrAmbiguousTopicInfo  = errors.New("could not identify info for topic")
	ErrNoRows              = errors.New("ensql: no rows in result set")
)

Standardized errors that the client may return from configuration issues or parsed from gRPC service calls. These errors can be evaluated using errors.Is to test for different error conditions in client code.

Functions

func Version added in v0.7.0 

func Version() string

Version returns the semantic version for the current build.

Types

type Acknowledger added in v0.7.1 

type Acknowledger interface {
	Ack(*api.Ack) error
	Nack(*api.Nack) error
}

Acknowledger allows consumers to send acks/nacks back to the server when they have successfully processed an event. An ack means that the event was processed and the consumer group offset can move on, while a nack means there was a local error and the nack code instructs the server how to handle the event. The subscriber implements this interface, but this can be mocked for testing events.

type Client 

type Client struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Client manages the credentials and connection to the Ensign server. The New() method creates a configured client and the Client methods are used to interact with the Ensign ecosystem, handling authentication, publish and subscribe streams, and interactions with topics. The Ensign client is the top-level method for creating Go applications that leverage data flows.

func New 

func New(opts ...Option) (client *Client, err error)

Create a new Ensign client, specifying connection and authentication options if necessary. Ensign expects that credentials are stored in the environment, set using the $ENSIGN_CLIENT_ID and $ENSIGN_CLIENT_SECRET environment variables. They can also be set manually using the WithCredentials or WithLoadCredentials options. You can also specify a mock ensign server to test your code that uses Ensign via WithMock. This function returns an error if the client is unable to dial ensign; however, authentication errors and connectivity checks may require an Ensign RPC call. You can use the Ping() method to check if your connection credentials to Ensign is correct.

func (*Client) ArchiveTopic added in v0.5.1 

func (c *Client) ArchiveTopic(ctx context.Context, topicID string) (_ api.TopicState, err error)

Archive a topic marking it as read-only.

func (*Client) Close 

func (c *Client) Close() (err error)

Close the connection to the current Ensign server. Closing the connection may block if streaming RPCs such as publish or subscribe are running. It is useful to Close the Ensign connection when you're done to free up any resources in long running programs, however, once closed, the Client cannot be reconnected and a new Client must be initialized to re-establish the connection.

func (*Client) ConnState added in v0.7.0 

func (c *Client) ConnState() connectivity.State

Conn state returns the connectivity state of the underlying gRPC connection.

Experimental: this method relies on an experimental gRPC API that could be changed.

func (*Client) CreateTopic added in v0.5.1 

func (c *Client) CreateTopic(ctx context.Context, topic string) (_ string, err error)

Create topic with the specified name and return the topic ID if there was no error. This method returns a gRPC error if the RPC cannot be successfully completed.

func (*Client) DestroyTopic added in v0.5.1 

func (c *Client) DestroyTopic(ctx context.Context, topicID string) (_ api.TopicState, err error)

Destroy a topic removing it and all of its data.

func (*Client) EnSQL added in v0.9.0 

func (c *Client) EnSQL(ctx context.Context, query *api.Query) (cursor *QueryCursor, err error)

EnSQL executes a query against Ensign and returns a cursor that can be used to fetch the event results. This RPC always returns a finite number of results. After all results have been returned the cursor will return nil. In order to retrieve events in a more streaming fashion, the Subscribe RPC should be used with a query option.

func (*Client) EnsignClient added in v0.5.1 

func (c *Client) EnsignClient() api.EnsignClient

Returns the underlying gRPC client for Ensign; useful for testing or advanced calls. It is not recommended to use this client for production code.

func (*Client) Explain added in v0.9.0 

func (c *Client) Explain(ctx context.Context, query *api.Query) (plan *api.QueryExplanation, err error)

Explain returns the query plan for the specified query, including the expected number of results and errors that might be returned.

func (*Client) Info added in v0.7.0 

func (c *Client) Info(ctx context.Context, topicIDs ...string) (info *api.ProjectInfo, err error)

Info returns summary statistics that describe the state of the project that you can connect to with your API key. Statistics include the number of topics, the number topics that are readonly (a subset of the number of topics) and the number of events. The project for the statistics is determined by the project your API key has access to (API keys are issued to projects). You can also specify a list of topicIDs to get the statistics for (e.g. filtering the statistics for one or more topics).

TODO: allow users to specify either topic names or topic IDs.

func (*Client) ListTopics added in v0.5.1 

func (c *Client) ListTopics(ctx context.Context) (topics []*api.Topic, err error)

ListTopics fetches all the topics that the client has access to in the project that the API keys are defined for. The ListTopics RPC is a paginated RPC, and this method continues to fetch all pages before returning a list of a results; fully materializing the list of topics in memory.

func (*Client) Publish 

func (c *Client) Publish(topic string, events ...*Event) (err error)

Publish one or more events to the specified topic name or topic ID. The first time that Publish is called, a Publisher stream is opened by the client that will run in its own go routine for the duration to the program; if the publish stream cannot be opened an error is returned. Otherwise, each event passed to the publish method will be sent to Ensign. If the Ensign connection has dropped or another connection error occurs an error will be returned. Once the event is published, it is up to the user to listen for an Ack or Nack on each event to determine if the event was specifically published or not.

func (*Client) PublishStream added in v0.7.0 

func (c *Client) PublishStream(ctx context.Context, opts ...grpc.CallOption) (api.Ensign_PublishClient, error)

PublishStream allows you to open a gRPC stream server to ensign for publishing API events directly. This manual mechanism of opening a stream is for advanced users and is not recommended in production. Instead using Publish or CreatePublisher is the best way to establish a stream connection to Ensign.

func (*Client) QuarterdeckClient added in v0.5.1 

func (c *Client) QuarterdeckClient() *auth.Client

Returns the underlying Quarterdeck authentication client; useful for testing or advanced calls. It is not recommended to use this client for production code.

func (*Client) SetTopicDeduplicationPolicy added in v0.11.0 

func (c *Client) SetTopicDeduplicationPolicy(ctx context.Context, topicID string, policy api.Deduplication_Strategy, offset api.Deduplication_OffsetPosition, keysOrFields []string, overwriteDuplicate bool) (_ api.TopicState, err error)

Set the topic deduplication policy on the server.

func (*Client) SetTopicShardingStrategy added in v0.11.0 

func (c *Client) SetTopicShardingStrategy(ctx context.Context, topicID string, strategy api.ShardingStrategy) (_ api.TopicState, err error)

Set the topic sharding strategy on the server.

func (*Client) Status added in v0.7.0 

func (c *Client) Status(ctx context.Context) (state *api.ServiceState, err error)

Status performs an unauthenticated check to the Ensign service to determine the state of the service. This may be useful in debugging connectivity issues.

TODO: update the return of status to include Quarterdeck status.

func (*Client) Subscribe 

func (c *Client) Subscribe(topics ...string) (sub *Subscription, err error)

Subscribe creates a subscription stream to the specified topics and returns a Subscription with a channel that can be listened on for incoming events. If the client cannot connect to Ensign or a subscription stream cannot be established, an error is returned.

func (*Client) SubscribeStream added in v0.7.0 

func (c *Client) SubscribeStream(ctx context.Context, opts ...grpc.CallOption) (api.Ensign_SubscribeClient, error)

SubscribeStream allows you to open a gRPC stream server to ensign for subscribing to API events directly. This manual mechanism of opening a stream is for advanced users and is not recommended in production. Instead using Subscribe or CreateSubscriber is the best way to establish a stream connection to Ensign.

func (*Client) TopicExists added in v0.5.1 

func (c *Client) TopicExists(ctx context.Context, topicName string) (_ bool, err error)

Check if a topic with the specified name exists in the project or not. The returned bool indicates if the topic exists; if an error is returned, then exists will be false. This method returns an gRPC error if the RPC cannot be successfully completed.

func (*Client) TopicID added in v0.5.1 

func (c *Client) TopicID(ctx context.Context, topicName string) (_ string, err error)

Find a topic ID from a topic name. TODO: automate and cache this on the client for easier lookups.

func (*Client) TopicInfo added in v0.9.0 

func (c *Client) TopicInfo(ctx context.Context, topicID ulid.ULID) (info *api.TopicInfo, err error)

func (*Client) WaitForConnStateChange added in v0.7.0 

func (c *Client) WaitForConnStateChange(ctx context.Context, sourceState connectivity.State) bool

Wait for the state of the underlying gRPC connection to change from the source state (not to the source state) or until the context times out. Returns true if the source state has changed to another state.

Experimental: this method relies on an experimental gRPC API that could be changed.

func (*Client) WaitForReconnect added in v0.7.0 

func (c *Client) WaitForReconnect(ctx context.Context) bool

WaitForReconnect checks if the connection has been reconnected periodically and returns true when the connection is ready. If the context deadline times out before a connection can be re-established, false is returned.

Experimental: this method relies on an experimental gRPC API that could be changed.

func (*Client) WithCallOptions added in v0.7.0 

func (c *Client) WithCallOptions(opts ...grpc.CallOption) *Client

WithCallOptions configures the next client Call to use the specified call options, after the call, the call options are removed. This method returns the Client pointer so that you can easily chain a call e.g. client.WithCallOptions(opts...).ListTopics() -- this ensures that we don't have to pass call options in to each individual call. Ensure that the clone of the client is discarded and garbage collected after use; the clone cannot be used to close the connection or fetch the options.

Experimental: call options and thread-safe cloning is an experimental feature and its signature may be subject to change in the future.

type Event added in v0.7.0 

type Event struct {
	// Metadata are user-defined key/value pairs that can be optionally added to an
	// event to store/lookup data without unmarshaling the entire payload.
	Metadata Metadata

	// Data is the datagram payload that defines the event.
	Data []byte

	// Mimetype describes how to parse the event datagram.
	Mimetype mimetype.MIME

	// Type defines the schema of the event datagram and is optional.
	Type *api.Type

	// Created is the timestamp that the event was created according to the client clock.
	Created time.Time
	// contains filtered or unexported fields
}

Events wrap user-defined datagrams that are totally ordered by the Ensign platform. Publishers create events with arbitrary data and send them to Ensign so that they can be sent to Subscribers awaiting the events or queried using EnSQL for later consumption. The datagram the event wraps is user-specific. It can be JSON, msgpack, text data, parquet, protocol buffers, etc. Applications should define event types using the Ensign schema registry and use those types to create events to publish and subscribe/query from.

func NewIncomingEvent added in v0.7.1 

func NewIncomingEvent(e *api.EventWrapper, sub Acknowledger) *Event

Creates a new incoming event as though it were from a subscription. This method is generally used by tests to crate mock events with an acknowledger for ensuring that an event is correctly acked/nacked to the consumer stream.

func NewOutgoingEvent added in v0.7.1 

func NewOutgoingEvent(e *api.EventWrapper, pub <-chan *api.PublisherReply) *Event

Creates a new outgoing event to be published. This method is generally used by tests to create mock events with the acked/nacked channels listening for a response from the publisher stream.

func (*Event) Ack added in v0.7.0 

func (e *Event) Ack() (bool, error)

Ack allows a user to acknowledge back to the Ensign server that an event received by a subscription stream has been successfully consumed. For consumer groups that have exactly-once or at-least-once semantics, this signals the message has been delivered successfully so as to not trigger a redelivery of the message to another consumer. Ack does not block and returns true if already acked. If a nack was sent before ack, then this method returns false. If this event was not received on a subscribe stream then an error is returned.

func (*Event) Acked added in v0.7.0 

func (e *Event) Acked() (bool, error)

Acked allows a user to check if an event published to an event stream has been successfully received by the server.

func (*Event) Clone added in v0.7.0 

func (e *Event) Clone() *Event

Clone the event, resetting its state and removing acks, nacks, created timestamp and context. Useful for resending events or for duplicating an event to edit and publish.

func (*Event) Committed added in v0.7.0 

func (e *Event) Committed() time.Time

Returns the committed timestamp if available.

func (*Event) Context added in v0.7.0 

func (e *Event) Context() context.Context

Context returns the message context if set otherwise a background context.

func (*Event) Equals added in v0.7.0 

func (e *Event) Equals(o *Event) bool

Compare two events to determine if they are equivalent by data. See Same() to determine if they are the same event by offset/topic.

func (*Event) Err added in v0.7.0 

func (e *Event) Err() error

Err returns any error that occurred processing the event.

func (*Event) ID added in v0.7.0 

func (e *Event) ID() string

Returns the event ID if the event has been published; otherwise returns empty string.

func (*Event) Info added in v0.7.0 

func (e *Event) Info() *api.EventWrapper

Returns the event wrapper which contains the API event info. Used for debugging.

func (*Event) LocalID added in v0.7.1 

func (e *Event) LocalID() []byte

Returns the topic ID that the event was published to if available; otherwise returns nil. This method is primarily for testing and debugging purposes; users should use the metadata to store application-specific ID material.

func (*Event) Nack added in v0.7.0 

func (e *Event) Nack(code api.Nack_Code) (bool, error)

Nack allows a user to signal to the Ensign server that an event received by a subscription stream has not been successfully consumed. For consumer groups that have exactly-once or at-least-once semantics, this signals the message needs to be redelivered to another consumer.

Nack does not block and returns true if already nacked. If an ack was sent before the nack, then this method returns false. If this event was not received on a subscribe stream then an error is returned.

func (*Event) Nacked added in v0.7.0 

func (e *Event) Nacked() (bool, error)

Nacked allows a user to check if an event published to an event stream has errored or otherwise been rejected by the server.

func (*Event) Offset added in v0.7.0 

func (e *Event) Offset() (offset uint64, epoch uint64)

Returns the offset and epoch of the event if available, otherwise returns 0.

func (*Event) Proto added in v0.7.1 

func (e *Event) Proto() *api.Event

Convert an event into a protocol buffer event.

func (*Event) SetContext added in v0.7.0 

func (e *Event) SetContext(ctx context.Context)

SetContext provides an event context for use in the handling application.

func (*Event) TopicID added in v0.7.0 

func (e *Event) TopicID() string

Returns the topic ID that the event was published to if available; otherwise returns an empty string. The TopicID is a ULID, the ULID can be parsed without going through a string representation using the TopicULID method. If the TopicID cannot be parsed as a ULID then a hexadecimal representation of the ID is returned. See the error from TopicULID for more info about what went wrong.

func (*Event) TopicULID added in v0.7.1 

func (e *Event) TopicULID() (topicID ulid.ULID, err error)

Returns the topic ULID that the event was published to if available, otherwise returns an error if there is no info, the topic ID is nil, or was unparseable.

type Metadata added in v0.7.0 

type Metadata map[string]string

Metadata are user-defined key/value pairs that can be optionally added to an event to store/lookup data without unmarshaling the entire payload.

func (Metadata) Get added in v0.7.0 

func (m Metadata) Get(key string) string

Get returns the metadata value for the given key. If the key is not in the metadata an empty string is returned without an error.

func (Metadata) Set added in v0.7.0 

func (m Metadata) Set(key, value string)

Set a metadata value for the given key; overwrites existing keys.

type NackError added in v0.7.0 

type NackError struct {
	ID      []byte
	Code    api.Nack_Code
	Message string
}

A Nack from the server on a publish stream indicates that the event was not successfully published for the reason specified by the code and the message. Nacks received by the publisher indicate that the event should be retried or dropped. Subscribers can also send NackErrors to the Ensign server in order to indicate that the message be replayed to a different client or that the consumer group offset should not be updated since the event was unhandled.

func (*NackError) Error added in v0.7.0 

func (e *NackError) Error() string

Error implements the error interface so that a NackError can be returned as an error.

type Option added in v0.7.0 

type Option func(o *Options) error

Option allows users to specify variadic options to create & connect the Ensign client.

func WithAuthenticator added in v0.7.0 

func WithAuthenticator(url string, noauth bool) Option

WithAuthenticator specifies a different Quarterdeck URL or you can supply an empty string and noauth set to true to have no authentication occur with the Ensign client.

func WithCredentials added in v0.7.0 

func WithCredentials(clientID, clientSecret string) Option

WithCredentials allows you to instantiate an Ensign client with API Key information.

func WithEnsignEndpoint added in v0.7.0 

func WithEnsignEndpoint(endpoint string, insecure bool, opts ...grpc.DialOption) Option

WithEnsignEndpoint allows you to specify an endpoint that is not the production Ensign cloud. This is useful if you're running an Ensign node in CI or connecting to a mock in local tests. Ensign developers may also use this to connect to staging. If any gRPC dial options are specified, they override the default Ensign dial options including the interceptors that perform authentication -- use only if you know what you're doing and why!

func WithLoadCredentials added in v0.7.0 

func WithLoadCredentials(path string) Option

WithLoadCredentials loads the Ensign API Key information from the JSON file that was download from the Rotational web application. Pass in the path to the credentials on disk to load them with this option!

func WithMock added in v0.7.0 

func WithMock(mock *mock.Ensign, opts ...grpc.DialOption) Option

WithMock connects ensign to the specified mock ensign server for local testing.

func WithOptions added in v0.7.0 

func WithOptions(opts Options) Option

WithOptions sets the options to the passed in options value. Note that this will override everything in the processing chain including zero-valued items; so use this as the first variadic option in NewOptions to guarantee correct processing.

type Options 

type Options struct {
	// The API Key credentials include the client ID and secret, both of which are
	// required to authenticate with Ensign via the authentication service so that an
	// access token can be retrieved and placed in all Ensign requests. The only time
	// these settings are not required is if NoAuthentication is true.
	ClientID     string
	ClientSecret string

	// The gRPC endpoint of the Ensign service; by default the EnsignEndpoint.
	Endpoint string

	// Dial options allows the user to specify gRPC connection options if necessary.
	// NOTE: use with care, this overrides the default dialing options including the
	// interceptors for authentication!
	Dialing []grpc.DialOption

	// The URL of the Quarterdeck system for authentication; by default AuthEndpoint.
	AuthURL string

	// If true, the client will not use TLS to connect to Ensign (default false).
	Insecure bool

	// If true, the client will not login with the api credentials and will omit access
	// tokens from Ensign RPCs. This is primarily used for testing against mocks.
	NoAuthentication bool

	// Mocking allows the client to be used in test code. Set testing mode to true and
	// create a *mock.Ensign to add to the dialer. Any other dialer options can also be
	// added to the mock for connection purposes.
	Testing bool
	Mock    *mock.Ensign
}

Options specifies the client configuration for authenticating and connecting to the Ensign service. The goal of the options struct is to be as minimal as possible. If users set their credentials via the environment, they should not have to specify any options at all to connect. The options does give the client flexibility to connect to Ensign nodes in other environments and is primarily for advanced usage.

func NewOptions added in v0.4.2 

func NewOptions(opts ...Option) (options Options, err error)

NewOptions instantiates an options object for configuring Ensign, sets defaults and loads missing options from the environment, then validates the options; returning an error if the options are incorrectly configured.

func (*Options) Validate 

func (o *Options) Validate() (err error)

Validate the options to make sure required configuration is set. This method also ensures that default values are set if a configuration is missing. For example, if the Endpoint is not set, this method first tries to set it from the environment, and then uses the default value as a last step.

type QueryCursor added in v0.9.0 

type QueryCursor struct {
	// contains filtered or unexported fields
}

QueryCursor exposes event results from an EnSQL query with familiar database cursor semantics. Note that the cursor is not thread safe and should only be used from a single thread.

func NewQueryCursor added in v0.9.0 

func NewQueryCursor(stream api.Ensign_EnSQLClient) (cursor *QueryCursor, err error)

NewQueryCursor creates a new query cursor that reads from the specified stream.

func (*QueryCursor) Close added in v0.9.0 

func (i *QueryCursor) Close() (err error)

Close the cursor, which closes the underlying stream.

func (*QueryCursor) FetchAll added in v0.9.0 

func (i *QueryCursor) FetchAll() (events []*Event, err error)

FetchAll returns all events from the query stream. If there are no more events then an empty slice is returned.

func (*QueryCursor) FetchMany added in v0.9.0 

func (i *QueryCursor) FetchMany(n int) (events []*Event, err error)

FetchMany returns the next n query results. If there are less than n results remaining then all the remaining results are returned.

func (*QueryCursor) FetchOne added in v0.9.0 

func (i *QueryCursor) FetchOne() (event *Event, err error)

FetchOne returns the next query result. If there are no more results then nil is returned.

type Subscription added in v0.7.0 

type Subscription struct {
	C <-chan *Event
	// contains filtered or unexported fields
}

A Subscription object with a channel of events is returned when you subscribe to a topic or topics. Listen on the provided channel in order to receive events from Ensign when they are published to your consumer group. It is the user's responsibility to Ack and Nack events when they are handled by using the methods on the event itself.

func (*Subscription) Close added in v0.7.0 

func (c *Subscription) Close() error

Close the subscription stream and associated channels, preventing any more events from being received and signaling to handler code that no more events will arrive.

Source Files

View all Source files

Directories

Path Synopsis
api
v1beta1
Package auth enables an Ensign client to authenticate with Ensign's Authn and Authz service called Quarterdeck.
 Package auth enables an Ensign client to authenticate with Ensign's Authn and Authz service called Quarterdeck.
authtest
Package authtest provides some simple JWT token testing functionality for use in Ensign SDK tests.
 Package authtest provides some simple JWT token testing functionality for use in Ensign SDK tests.
mimetype
v1beta1
Package mock implements an in-memory gRPC mock Ensign server that can be connected to using a bufconn.
 Package mock implements an in-memory gRPC mock Ensign server that can be connected to using a bufconn.
region
v1beta1
Package stream provides
 Package stream provides

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.