connector

package
v1.45.0 Latest Latest
Warning

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

 Go to latest Published: Jul 7, 2026 License: Apache-2.0 Imports: 75 Imported by: 0

Documentation

Overview

Package connector implements the base class of all microservices.

Index

Constants

View Source 
const (
	PROD    string = "PROD"    // PROD for a production environment
	LAB     string = "LAB"     // LAB for all non-production environments such as dev integration, test, staging, etc.
	LOCAL   string = "LOCAL"   // LOCAL when developing on the local machine
	TESTING string = "TESTING" // TESTING when running inside a testing app
)

Deployment environments

View Source 
const (
	Gray    = "\033[38;2;128;128;128m" // #808080
	Magenta = "\033[38;2;197;134;192m" // #c586c0
	Yellow  = "\033[38;2;215;186;125m" // #d7ba7d
	Orange  = "\033[38;2;206;145;120m" // #ce9178
	Green   = "\033[38;2;106;153;85m"  // #6A9955
	Cyan    = "\033[38;2;156;220;254m" // #9cdcfe
	Red     = "\033[38;2;244;71;71m"   // #f44747
	White   = "\033[38;2;212;212;212m" // #d4d4d4
	Blue    = "\033[38;2;86;159;214m"  // #569cd6
	Reset   = "\033[0m"
)

Variables

This section is empty.

Functions

func SubjectOfRequest added in v1.29.0 

func SubjectOfRequest(plane, port, srcHostname, hostname, idOrLocality, method, path string) string

SubjectOfRequest is the subject a microservice publishes an outgoing request to. Argument order mirrors the on-wire segment order. Exposed alongside SubjectOfRequestSub for cross-tool wire-format verification.

func SubjectOfRequestSub added in v1.29.0 

func SubjectOfRequestSub(plane, port, hostname, idOrLocality, method, path string) string

SubjectOfRequestSub is the subject a microservice subscribes to in order to receive incoming requests for a given route. Source is wildcarded; port 0 and method "ANY" become wildcards. Exposed as the runtime's contract: external tools (notably cmd/genacl) compute their NATS subject patterns to match this exact output.

Types

type Connector 

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

Connector is the base class of a microservice. It provides the microservice such functions as connecting to the NATS messaging bus, communications with other microservices, logging, config, etc.

func New 

func New(hostname string) *Connector

New constructs a new Connector with the given hostname.

func NewConnector 

func NewConnector() *Connector

NewConnector constructs a new Connector.

func (*Connector) ActivateSubscription added in v1.30.0 

func (c *Connector) ActivateSubscription(name string) error

ActivateSubscription activates the named subscription if it is currently off-bus. The subscription joins the transport and a peer notification is broadcast so other microservices invalidate their cached known-responders entries. Already-active subscriptions are silently skipped, so the call is idempotent. Valid while the connector is starting up, started, or shutting down - this lets sub.Manual subscriptions come on/off bus alongside the lifecycle of their backing resource (e.g. a Python venv allocated inside OnStartup, or a distributed cache torn down inside OnShutdown). Returns an error if no subscription with that name is registered.

func (*Connector) Config 

func (c *Connector) Config(name string) (value string)

Config returns the value of a previously defined property. The value of the property is available after the microservice has started after being obtained from the configurator microservice. Config property names are case sensitive.

func (*Connector) DeactivateSubscription added in v1.30.0 

func (c *Connector) DeactivateSubscription(name string) error

DeactivateSubscription takes the named subscription off the bus while leaving it registered in the connector, so a subsequent Connector.ActivateSubscription brings it back online. Unicast callers see a clean 404 ack-timeout while it is deactivated; load-balancing routes around the cold replica. Already-deactivated subscriptions are silently skipped, so the call is idempotent. Valid while the connector is starting up, started, or shutting down. Returns an error if no subscription with that name is registered.

func (*Connector) DefineConfig 

func (c *Connector) DefineConfig(name string, options ...cfg.Option) error

DefineConfig defines a property used to configure the microservice. Properties must be defined before the service starts. Config property names are case sensitive.

func (*Connector) Deployment 

func (c *Connector) Deployment() string

Deployment indicates what deployment environment the microservice is running in: PROD for a production environment; LAB for all non-production environments such as dev integration, test, staging, etc.; LOCAL when developing on the local machine; TESTING when running inside a testing app.

func (*Connector) DescribeCounter added in v1.13.1 

func (c *Connector) DescribeCounter(name string, desc string) (err error)

DescribeCounter defines a new counter metric.

func (*Connector) DescribeGauge added in v1.13.1 

func (c *Connector) DescribeGauge(name string, desc string) (err error)

DescribeGauge defines a new gauge metric.

func (*Connector) DescribeHistogram added in v1.13.1 

func (c *Connector) DescribeHistogram(name string, desc string, bucketBounds []float64) (err error)

DescribeHistogram defines a new histogram metric.

func (*Connector) Description 

func (c *Connector) Description() string

Description returns the human-friendly description of the microservice.

func (*Connector) DistribCache 

func (c *Connector) DistribCache() *dlru.Cache

DistribCache is a cache that stores data among all peers of the microservice. By default the cache is limited to 32MB per peer and a 1 hour TTL.

Operating on a distributed cache is slower than on a local cache because it involves network communication among peers. However, the memory capacity of a distributed cache scales linearly with the number of peers and its content is often able to survive a restart of the microservice.

Cache elements can get evicted for various reason and without warning. Cache only that which you can afford to lose and reconstruct. Do not use the cache to share state among peers. The cache is subject to race conditions in rare situations.

func (*Connector) ExternalizeURL added in v1.27.0 

func (c *Connector) ExternalizeURL(ctx context.Context, internalURL string) string

ExternalizeURL converts an internal Microbus URL to an external one using the X-Forwarded-* headers from the context's frame. The internal Microbus URL is assumed relative to the hostname of this connector. Relative URLs are returned as is.

Examples:

func (*Connector) ForceTrace 

func (c *Connector) ForceTrace(ctx context.Context)

ForceTrace forces the trace containing the span to be exported

func (*Connector) GET 

func (c *Connector) GET(ctx context.Context, url string) (*http.Response, error)

GET makes a GET request.

func (*Connector) Go 

func (c *Connector) Go(ctx context.Context, f func(ctx context.Context) (err error)) error

Go launches a goroutine in the lifetime context of the microservice. Errors and panics are automatically captured and logged. On shutdown, the microservice will attempt to gracefully end a pending goroutine before termination.

func (*Connector) Hostname 

func (c *Connector) Hostname() string

Hostname returns the hostname of the microservice. A microservice is addressable by its hostname.

func (*Connector) ID 

func (c *Connector) ID() string

ID is a unique identifier of a particular instance of the microservice

func (*Connector) IncrementCounter added in v1.22.0 

func (c *Connector) IncrementCounter(ctx context.Context, name string, val float64, attributes ...any) (err error)

IncrementCounter adds a non-negative value to a counter metric. Attributes conform to the standard slog pattern.

func (*Connector) Init added in v1.17.0 

func (c *Connector) Init(initializer func(c *Connector) (err error)) *Connector

Init enables a single-statement pattern for initializing the connector.

func (*Connector) IsStarted 

func (c *Connector) IsStarted() bool

IsStarted indicates if the microservice has been successfully started up.

func (*Connector) Lifetime 

func (c *Connector) Lifetime() context.Context

Lifetime returns a context that becomes valid before OnStartup runs and is cancelled only after OnShutdown returns and the connector's soft drain elapses. It is the canonical root context for long-lived goroutines launched from OnStartup (worker pools, refillers, background reconcilers) and for outbound calls that should outlive the request that initiated them. Its Done channel signals "the microservice is shutting down, finish up." Use it instead of context.Background for any work whose lifecycle should track the microservice's.

func (*Connector) LoadResString 

func (c *Connector) LoadResString(ctx context.Context, stringKey string) (value string, err error)

LoadResString returns a localized string from the string bundle best matched to the locale of the context. The string bundle is a YAML file that must be loadable from the service's resource FS with the name text.yaml. The YAML should map a string key to its localized values on a per language basis: 

Localized:
  en: Localized
  en-UK: Localised
  fr: Localisée
  it: Localizzato
  default: Localized

If a default is not explicitly provided, English (en) is used as the fallback language. String keys and ISO 639 language codes are case sensitive.

func (*Connector) LoadResStrings added in v1.18.0 

func (c *Connector) LoadResStrings(ctx context.Context) (valuesByStringKey map[string]string, err error)

LoadResStrings returns all strings from the string bundle best matched to the locale in the context.

func (*Connector) Locality 

func (c *Connector) Locality() string

Locality returns the geographic locality of the microservice.

func (*Connector) LogDebug 

func (c *Connector) LogDebug(ctx context.Context, msg string, args ...any)

LogDebug logs a message at DEBUG level. DEBUG level messages are ignored in PROD environments or if the MICROBUS_LOG_DEBUG environment variable is not set. The message should be static and concise. Optional arguments can be added for variable data. Arguments conform to the standard slog pattern.

Example: 

c.LogDebug(ctx, "Tight loop", "index", i)

func (*Connector) LogError 

func (c *Connector) LogError(ctx context.Context, msg string, args ...any)

LogError logs a message at ERROR level. The message should be static and concise. Optional arguments can be added for variable data. Arguments conform to the standard slog pattern. When logging an error object, name it "error".

Example: 

c.LogError(ctx, "Opening file", "error", err, "file", fileName)

func (*Connector) LogInfo 

func (c *Connector) LogInfo(ctx context.Context, msg string, args ...any)

LogInfo logs a message at INFO level. The message should be static and concise. Optional arguments can be added for variable data. Arguments conform to the standard slog pattern.

Example: 

c.LogInfo(ctx, "File uploaded", "gb", sizeGB)

func (*Connector) LogWarn 

func (c *Connector) LogWarn(ctx context.Context, msg string, args ...any)

LogWarn logs a message at WARN level. The message should be static and concise. Optional arguments can be added for variable data. Arguments conform to the standard slog pattern.

Example: 

c.LogWarn(ctx, "Dropping job", "job", jobID)

func (*Connector) Logger added in v1.38.0 

func (c *Connector) Logger() *slog.Logger

Logger returns the structured logger of the microservice. Before Startup it returns a logger that discards all records, matching the no-op behavior of LogDebug/LogInfo/LogWarn/LogError. Logging through the logger's context-aware methods (DebugContext, InfoContext, WarnContext, ErrorContext) enriches each record with trace correlation and actor identity drawn from the context; the non-context methods (Debug, Info, ...) log without that enrichment. The convenience LogDebug/LogInfo/LogWarn/LogError methods delegate to the context-aware methods.

Fetch the logger at point of use rather than caching it across Startup: the discard logger is replaced by the real one during Startup, so a reference held from before Startup keeps discarding.

func (*Connector) MeterProvider added in v1.38.0 

func (c *Connector) MeterProvider() metric.MeterProvider

MeterProvider returns the OpenTelemetry meter provider of the microservice, or a no-op provider when metrics are not configured. Use it to instrument third-party libraries against the same pipeline and resource as the framework. Every instrument it vends stamps the microservice's identity attributes (service, deployment, plane, ver, id) onto each measurement, so a library's metrics carry the same labels as the framework's own microbus_* metrics and are filterable by service and deployment without a target_info join.

func (*Connector) MustLoadResString added in v1.18.0 

func (c *Connector) MustLoadResString(ctx context.Context, stringKey string) string

MustLoadResString returns a string from the string bundle. It panics if the string is not found.

func (*Connector) MustLoadResStrings added in v1.18.0 

func (c *Connector) MustLoadResStrings(ctx context.Context) (valuesByStringKey map[string]string)

MustLoadResStrings returns all strings from from the string bundle. It panics if the string bundle is not found.

func (*Connector) MustReadResFile 

func (c *Connector) MustReadResFile(name string) []byte

MustReadResFile returns the content of a resource file, or nil if not found. It panics if the resource file is not found.

func (*Connector) MustReadResTextFile 

func (c *Connector) MustReadResTextFile(name string) string

MustReadResTextFile returns the content of a resource file as a string, or "" if not found. It panics if the resource file is not found.

func (*Connector) POST 

func (c *Connector) POST(ctx context.Context, url string, body any) (*http.Response, error)

POST makes a POST request. Body of type io.Reader, []byte and string is serialized in binary form. url.Values is serialized as form data. All other types are serialized as JSON.

func (*Connector) Parallel 

func (c *Connector) Parallel(jobs ...func() (err error)) error

Parallel executes multiple jobs in parallel and returns the first error it encounters. It is a convenient pattern for calling multiple other microservices and thus amortize the network latency. There is no mechanism to identify the failed jobs so this pattern isn't suited for jobs that update data and require to be rolled back on failure.

func (*Connector) Plane 

func (c *Connector) Plane() string

Plane is a unique prefix set for all communications sent or received by this microservice. It is used to isolate communication among a group of microservices over a NATS cluster that is shared with other microservices. If not explicitly set, the value is pulled from the Plane config, or the default "microbus" is used

func (*Connector) Publish 

func (c *Connector) Publish(ctx context.Context, options ...pub.Option) iter.Seq[*pub.Response]

Publish makes an HTTP request then awaits and returns the responses asynchronously. By default, publish performs a multicast and multiple responses (or none at all) may be returned. Use the Request method or pass in pub.Unicast() to Publish to perform a unicast.

func (*Connector) ReadResFile 

func (c *Connector) ReadResFile(name string) ([]byte, error)

ReadResFile returns the content of a resource file.

func (*Connector) ReadResTextFile 

func (c *Connector) ReadResTextFile(name string) (string, error)

ReadResTextFile returns the content of a resource file as a string.

func (*Connector) RecordGauge added in v1.13.1 

func (c *Connector) RecordGauge(ctx context.Context, name string, val float64, attributes ...any) (err error)

RecordGauge observes a value for a gauge metric. Attributes conform to the standard slog pattern.

func (*Connector) RecordHistogram added in v1.13.1 

func (c *Connector) RecordHistogram(ctx context.Context, name string, val float64, attributes ...any) (err error)

RecordHistogram observes a value for a histogram metric. Attributes conform to the standard slog pattern.

func (*Connector) Request 

func (c *Connector) Request(ctx context.Context, options ...pub.Option) (*http.Response, error)

Request makes an HTTP request then awaits and returns a single response synchronously. If no response is received, an ack timeout (404) error is returned.

func (*Connector) ResFS 

func (c *Connector) ResFS() fs.FS

ResFS returns the FS associated with the connector.

func (*Connector) ResetConfig 

func (c *Connector) ResetConfig(name string) error

ResetConfig resets the value of a previously defined configuration property to its default value. This action is restricted to the TESTING deployment in which the fetching of values from the configurator is disabled. Config property names are case sensitive.

func (*Connector) ServeResFile 

func (c *Connector) ServeResFile(name string, w http.ResponseWriter, r *http.Request) error

ServeResFile serves the content of a resources file as a response to a web request.

func (*Connector) SetConfig 

func (c *Connector) SetConfig(name string, value any) error

SetConfig sets the value of a previously defined configuration property. This action is restricted to the TESTING deployment in which the fetching of values from the configurator is disabled. Config property names are case sensitive.

func (*Connector) SetDeployment 

func (c *Connector) SetDeployment(deployment string) error

SetDeployment sets what deployment environment the microservice is running in. Explicitly setting a deployment will override any value specified by the Deployment config property. Setting an empty value will clear this override.

Valid values are: PROD for a production environment; LAB for all non-production environments such as dev integration, test, staging, etc.; LOCAL when developing on the local machine; TESTING when running inside a testing app.

func (*Connector) SetDescription 

func (c *Connector) SetDescription(description string) error

SetDescription sets a human-friendly description of the microservice.

func (*Connector) SetHostname 

func (c *Connector) SetHostname(hostname string) error

SetHostname sets the hostname of the microservice. The hostname must be a canonical Microbus identity per httpx.ValidateHostname: lowercase letters, digits, dots, and hyphens; no underscores, no "id-" or "loc-" first segment, not "all" or "*.all", no leading/trailing whitespace. For example, this.is.a.valid.host-name.123.local

func (*Connector) SetLocality 

func (c *Connector) SetLocality(locality string) error

SetLocality sets the geographic locality of the microservice which is used to optimize routing. Localities are hierarchical with the broadest identifier first, separated by hyphens, similar to AWS region/AZ identifiers such as "us-west-b-1" or arbitrarily "europe-italy-rome". DNS-style dot notation with the most specific identifier first is also accepted: "1.b.west.us" is equivalent to "us-west-b-1". Localities are case-insensitive. Letters, numbers, hyphens and underscores are allowed. The special values "AWS" or "GCP" can be set to determine the locality automatically from the cloud provider's meta-data servers.

func (*Connector) SetOnConfigChanged 

func (c *Connector) SetOnConfigChanged(handler service.ConfigChangedHandler) error

SetOnConfigChanged sets the function to be called when a new config was received from the configurator.

func (*Connector) SetOnObserveMetrics added in v1.13.1 

func (c *Connector) SetOnObserveMetrics(handler service.ObserveMetricsHandler) error

SetOnObserveMetrics adds a function to be called just before metrics are produced to allow observing them just in time.

func (*Connector) SetOnShutdown 

func (c *Connector) SetOnShutdown(handler service.ShutdownHandler) error

SetOnShutdown sets the function to be called during the shutting down of the microservice.

func (*Connector) SetOnStartup 

func (c *Connector) SetOnStartup(handler service.StartupHandler) error

SetOnStartup sets the function to be called during the starting up of the microservice.

func (*Connector) SetPlane 

func (c *Connector) SetPlane(plane string) error

SetPlane sets a unique prefix for all communications sent or received by this microservice. A plane is used to isolate communication among a group of microservices over a NATS cluster that is shared with other microservices. Explicitly setting a plane overrides any value specified by the Plane config. The plane can only contain alphanumeric case-sensitive characters. Setting an empty value will clear this override

func (*Connector) SetResFS 

func (c *Connector) SetResFS(resFileSys fs.FS) error

SetResFS initialized the connector to load resource files from an arbitrary FS.

func (*Connector) SetResFSDir 

func (c *Connector) SetResFSDir(directoryPath string) error

SetResFSDir initialized the connector to load resource files from a directory.

func (*Connector) SetVersion 

func (c *Connector) SetVersion(version int) error

SetVersion sets the sequential version number of the microservice.

func (*Connector) Shutdown 

func (c *Connector) Shutdown(ctx context.Context) (err error)

Shutdown the microservice by deactivating subscriptions and disconnecting from the transport. The deadline on ctx, if any, bounds the time allotted to the operation: OnShutdown receives the remaining budget, then the drains and teardown share what's left.

func (*Connector) Sleep added in v1.19.0 

func (c *Connector) Sleep(ctx context.Context, duration time.Duration) error

Sleep pauses the current goroutine for the specified duration, or until the provided context or the lifetime context of the microservice is canceled or its deadline is exceeded. It returns nil if the full duration elapsed, or the canceling context's error (context.Canceled or context.DeadlineExceeded), traced, if interrupted.

func (*Connector) Span 

func (c *Connector) Span(ctx context.Context) trc.Span

Span returns the tracing span stored in the context.

func (*Connector) StartSpan 

func (c *Connector) StartSpan(ctx context.Context, spanName string, opts ...trc.Option) (context.Context, trc.Span)

StartSpan creates a tracing span and a context containing the newly-created span. If the context provided already contains asSpan then the newly-created span will be a child of that span, otherwise it will be a root span.

Any Span that is created must also be ended. This is the responsibility of the user. Implementations of this API may leak memory or other resources if spans are not ended.

func (*Connector) StartTicker 

func (c *Connector) StartTicker(name string, interval time.Duration, handler service.TickerHandler) error

StartTicker initiates a recurring job at a set interval. Tickers do not run when the connector is running in the TESTING deployment environment. Ticker names are case sensitive.

func (*Connector) Startup 

func (c *Connector) Startup(ctx context.Context) (err error)

Startup the microservice by connecting to the transport and activating the subscriptions.

func (*Connector) StopTicker added in v1.5.0 

func (c *Connector) StopTicker(name string) error

StopTicker stops a running ticker. Ticker names are case-insensitive. Ticker names are case sensitive.

func (*Connector) Subscribe 

func (c *Connector) Subscribe(name string, handler sub.HTTPHandler, options ...sub.Option) error

Listen subscribes a handler under a typed name. The name must be a Go-style upper-case identifier (e.g. "MyEndpoint2") and unique within the connector. Exactly one feature option (sub.Function, sub.Web, sub.InboundEvent, sub.Task, sub.Workflow) must be supplied.

Defaults filled in after options are applied:

  • Method defaults to "ANY".
  • Route defaults to ":443/my-subscription" with the port determined by the feature type (443 for function/web, 417 for inbound events, 428 for tasks/graphs).
  • Queue defaults to the connector's hostname.

Returns an error if the name is invalid, already registered, or if the option set is malformed.

func (*Connector) Subscriptions added in v1.30.0 

func (c *Connector) Subscriptions() []SubscriptionInfo

Subscriptions returns a read-only snapshot of every subscription registered with the connector, in registration order. Callers iterate the result and act on names of interest via Connector.ActivateSubscription and Connector.DeactivateSubscription. Filter by any combination of fields (SubscriptionInfo.Tags, SubscriptionInfo.Type, etc.) to express "all Python tasks" or "all manual subs" without a custom query API.

func (*Connector) TracerProvider added in v1.38.0 

func (c *Connector) TracerProvider() trace.TracerProvider

TracerProvider returns the OpenTelemetry tracer provider of the microservice, or a no-op provider when tracing is not configured. Use it to instrument third-party libraries against the same pipeline and resource as the framework.

func (*Connector) Unsubscribe 

func (c *Connector) Unsubscribe(name string) error

Unsubscribe removes the subscription registered with Connector.Subscribe under the given name. Returns an error if no subscription with that name exists.

func (*Connector) Version 

func (c *Connector) Version() int

Version is the sequential version number of the microservice.

func (*Connector) WriteResTemplate added in v1.21.0 

func (c *Connector) WriteResTemplate(w io.Writer, name string, data any) (err error)

WriteResTemplate parses the resource file as a template, executes it given the data, writing the result to the writer.

The template is assumed to be a text template unless the file name ends in .html, in which case it is processed as an HTML template. {{ var | attr }}, {{ var | url }}, {{ var | css }} or {{ var | safe }} may be used to prevent the escaping of a variable in an HTML template. These map to htmltemplate.HTMLAttr, htmltemplate.URL, htmltemplate.CSS and htmltemplate.HTML respectively. Use of these types presents a security risk.

This method does not support customizing execution with a func map or changing the delimiters. If either is required, use the standard library pattern instead.

type HTTPHandler 

type HTTPHandler = sub.HTTPHandler

HTTPHandler extends the standard http.Handler to also return an error

type SubscriptionInfo added in v1.30.0 

type SubscriptionInfo struct {
	Name           string
	Description    string
	Host           string
	Port           string
	Method         string
	Path           string
	Queue          string
	Type           string
	RequiredClaims string
	Tags           []string
	Manual         bool
	NoTrace        bool
	Active         bool
}

SubscriptionInfo is a read-only snapshot of one subscription as it appears in the connector.

Source Files

View all Source files

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.