Archived
Vendor every dependency, so no build fetches the host's validator (hq to-be 45 D1)
The controller imports mesh-host/validate through a replace onto the forge that holds it, and every build — the build agent's go build in a fresh toolchain container, the Dockerfile's go mod download — would have fetched it through the public proxy and checksum database at build time: a merge breaking main on the network, the class Phase 1 removes. vendor/ is committed; go builds from it with nothing fetched, and refuses to build when it and go.mod disagree, so a pin moved without go mod vendor fails at once. The Dockerfile copies vendor/ and builds with GOPROXY=off.
This commit is contained in:
jochen
+303
@@ -0,0 +1,303 @@
|
||||
# NATS micro [](https://pkg.go.dev/github.com/nats-io/nats.go/micro)
|
||||
|
||||
- [Overview](#overview)
|
||||
- [Basic usage](#basic-usage)
|
||||
- [Endpoints and groups](#endpoints-and-groups)
|
||||
- [Discovery and Monitoring](#discovery-and-monitoring)
|
||||
- [Examples](#examples)
|
||||
- [Documentation](#documentation)
|
||||
|
||||
## Overview
|
||||
|
||||
The `micro` package in the NATS.go library provides a simple way to create
|
||||
microservices that leverage NATS for scalability, load management and
|
||||
observability.
|
||||
|
||||
## Basic usage
|
||||
|
||||
To start using the `micro` package, import it in your application:
|
||||
|
||||
```go
|
||||
import "github.com/nats-io/nats.go/micro"
|
||||
```
|
||||
|
||||
The core of the `micro` package is the Service. A Service aggregates endpoints
|
||||
for handling application logic. Services are named and versioned. You create a
|
||||
Service using the `micro.AddService()` function, passing in the NATS connection
|
||||
and Service configuration.
|
||||
|
||||
```go
|
||||
nc, _ := nats.Connect(nats.DefaultURL)
|
||||
|
||||
// request handler
|
||||
echoHandler := func(req micro.Request) {
|
||||
req.Respond(req.Data())
|
||||
}
|
||||
|
||||
srv, err := micro.AddService(nc, micro.Config{
|
||||
Name: "EchoService",
|
||||
Version: "1.0.0",
|
||||
// base handler
|
||||
Endpoint: µ.EndpointConfig{
|
||||
Subject: "svc.echo",
|
||||
Handler: micro.HandlerFunc(echoHandler),
|
||||
},
|
||||
})
|
||||
```
|
||||
|
||||
After creating the service, it can be accessed by publishing a request on
|
||||
endpoint subject. For given configuration, run:
|
||||
|
||||
```sh
|
||||
nats req svc.echo "hello!"
|
||||
```
|
||||
|
||||
To get:
|
||||
|
||||
```sh
|
||||
17:37:32 Sending request on "svc.echo"
|
||||
17:37:32 Received with rtt 365.875µs
|
||||
hello!
|
||||
```
|
||||
|
||||
## Endpoints and groups
|
||||
|
||||
Base endpoint can be optionally configured on a service, but it is also possible
|
||||
to add more endpoints after the service is created.
|
||||
|
||||
```go
|
||||
srv, _ := micro.AddService(nc, config)
|
||||
|
||||
// endpoint will be registered under "svc.add" subject
|
||||
err = srv.AddEndpoint("svc.add", micro.HandlerFunc(add))
|
||||
```
|
||||
|
||||
In the above example `svc.add` is an endpoint name and subject. It is possible
|
||||
have a different endpoint name then the endpoint subject by using
|
||||
`micro.WithEndpointSubject()` option in `AddEndpoint()`.
|
||||
|
||||
```go
|
||||
// endpoint will be registered under "svc.add" subject
|
||||
err = srv.AddEndpoint("Adder", micro.HandlerFunc(echoHandler), micro.WithEndpointSubject("svc.add"))
|
||||
```
|
||||
|
||||
Endpoints can also be aggregated using groups. A group represents a common
|
||||
subject prefix used by all endpoints associated with it.
|
||||
|
||||
```go
|
||||
srv, _ := micro.AddService(nc, config)
|
||||
|
||||
numbersGroup := srv.AddGroup("numbers")
|
||||
|
||||
// endpoint will be registered under "numbers.add" subject
|
||||
_ = numbersGroup.AddEndpoint("add", micro.HandlerFunc(addHandler))
|
||||
// endpoint will be registered under "numbers.multiply" subject
|
||||
_ = numbersGroup.AddEndpoint("multiply", micro.HandlerFunc(multiplyHandler))
|
||||
```
|
||||
|
||||
## Customizing queue groups
|
||||
|
||||
For each service, group and endpoint the queue group used to gather responses
|
||||
can be customized or disabled. If not provided a default queue group will be
|
||||
used (`q`). Customizing queue groups can be useful to e.g. implement fanout
|
||||
request pattern or hedged request pattern (to reduce tail latencies by only
|
||||
waiting for the first response for multiple service instances).
|
||||
|
||||
Let's say we have multiple services listening on the same subject, but with
|
||||
different queue groups:
|
||||
|
||||
```go
|
||||
for i := 0; i < 5; i++ {
|
||||
srv, _ := micro.AddService(nc, micro.Config{
|
||||
Name: "EchoService",
|
||||
Version: "1.0.0",
|
||||
QueueGroup: fmt.Sprintf("q-%d", i),
|
||||
// base handler
|
||||
Endpoint: µ.EndpointConfig{
|
||||
Subject: "svc.echo",
|
||||
Handler: micro.HandlerFunc(echoHandler),
|
||||
},
|
||||
})
|
||||
}
|
||||
```
|
||||
|
||||
In the client, we can send request to `svc.echo` to receive responses from all
|
||||
services registered on this subject (or wait only for the first response):
|
||||
|
||||
```go
|
||||
sub, _ := nc.SubscribeSync("rply")
|
||||
nc.PublishRequest("svc.echo", "rply", nil)
|
||||
for start := time.Now(); time.Since(start) < 5*time.Second; {
|
||||
msg, err := sub.NextMsg(1 * time.Second)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
fmt.Println("Received ", string(msg.Data))
|
||||
}
|
||||
```
|
||||
|
||||
Queue groups can be overwritten by setting them on groups and endpoints as well:
|
||||
|
||||
```go
|
||||
srv, _ := micro.AddService(nc, micro.Config{
|
||||
Name: "EchoService",
|
||||
Version: "1.0.0",
|
||||
QueueGroup: "q1",
|
||||
})
|
||||
|
||||
g := srv.AddGroup("g", micro.WithGroupQueueGroup("q2"))
|
||||
|
||||
// will be registered with queue group 'q2' from parent group
|
||||
g.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}))
|
||||
|
||||
// will be registered with queue group 'q3'
|
||||
g.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}), micro.WithEndpointQueueGroup("q3"))
|
||||
```
|
||||
|
||||
Similarly, queue groups can be disabled on service config, group and endpoint levels. If disabled,
|
||||
a standard NATS subscription will be created for the endpoint.
|
||||
|
||||
```go
|
||||
// disable queue group for the service
|
||||
srv, _ := micro.AddService(nc, micro.Config{
|
||||
Name: "EchoService",
|
||||
Version: "1.0.0",
|
||||
QueueGroupDisabled: true,
|
||||
})
|
||||
|
||||
// create a group with queue group disabled
|
||||
srv.AddGroup("g", micro.WithEndpointQueueGroupDisabled())
|
||||
|
||||
// create an endpoint with queue group disabled
|
||||
srv.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}), micro.WithEndpointQueueGroupDisabled())
|
||||
```
|
||||
|
||||
When disabling queue groups, same inheritance rules apply as for customizing
|
||||
queue groups. (service config -> group -> endpoint)
|
||||
|
||||
## Discovery and Monitoring
|
||||
|
||||
Each service is assigned a unique ID on creation. A service instance is
|
||||
identified by service name and ID. Multiple services with the same name, but
|
||||
different IDs can be created.
|
||||
|
||||
Each service exposes 3 endpoints when created:
|
||||
|
||||
- PING - used for service discovery and RTT calculation
|
||||
- INFO - returns service configuration details (used subjects, service metadata
|
||||
etc.)
|
||||
- STATS - service statistics
|
||||
|
||||
Each of those operations can be performed on 3 subjects:
|
||||
|
||||
- all services: `$SRV.<operation>` - returns a response for each created service
|
||||
and service instance
|
||||
- by service name: `$SRV.<operation>.<service_name>` - returns a response for
|
||||
each service with given `service_name`
|
||||
- by service name and ID: `$SRV.<operation>.<service_name>.<service_id>` -
|
||||
returns a response for a service with given `service_name` and `service_id`
|
||||
|
||||
For given configuration
|
||||
|
||||
```go
|
||||
nc, _ := nats.Connect("nats://localhost:4222")
|
||||
echoHandler := func(req micro.Request) {
|
||||
req.Respond(req.Data())
|
||||
}
|
||||
|
||||
config := micro.Config{
|
||||
Name: "EchoService",
|
||||
Version: "1.0.0",
|
||||
Endpoint: µ.EndpointConfig{
|
||||
Subject: "svc.echo",
|
||||
Handler: micro.HandlerFunc(echoHandler),
|
||||
},
|
||||
}
|
||||
for i := 0; i < 3; i++ {
|
||||
srv, err := micro.AddService(nc, config)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer srv.Stop()
|
||||
}
|
||||
```
|
||||
|
||||
Service IDs can be discovered by:
|
||||
|
||||
```sh
|
||||
nats req '$SRV.PING.EchoService' '' --replies=3
|
||||
|
||||
13:03:04 Sending request on "$SRV.PING.EchoService"
|
||||
13:03:04 Received with rtt 1.302208ms
|
||||
{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4K7","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
|
||||
|
||||
13:03:04 Received with rtt 1.317ms
|
||||
{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Kt","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
|
||||
|
||||
13:03:04 Received with rtt 1.320291ms
|
||||
{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Lf","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
|
||||
```
|
||||
|
||||
A specific service instance info can be retrieved:
|
||||
|
||||
```sh
|
||||
nats req '$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
|
||||
|
||||
13:04:19 Sending request on "$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
|
||||
13:04:19 Received with rtt 318.875µs
|
||||
{
|
||||
"name": "EchoService",
|
||||
"id": "x3Yuiq7g7MoxhXdxk7i4K7",
|
||||
"version": "1.0.0",
|
||||
"metadata": {},
|
||||
"type": "io.nats.micro.v1.info_response",
|
||||
"description": "",
|
||||
"endpoints": [
|
||||
{
|
||||
"name": "default",
|
||||
"subject": "svc.echo",
|
||||
"queue_group": "q",
|
||||
"metadata": null
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
To get statistics for this service:
|
||||
|
||||
```sh
|
||||
nats req '$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
|
||||
|
||||
13:04:46 Sending request on "$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
|
||||
13:04:46 Received with rtt 678.25µs
|
||||
{
|
||||
"name": "EchoService",
|
||||
"id": "x3Yuiq7g7MoxhXdxk7i4K7",
|
||||
"version": "1.0.0",
|
||||
"metadata": {},
|
||||
"type": "io.nats.micro.v1.stats_response",
|
||||
"started": "2024-09-24T11:02:55.564771Z",
|
||||
"endpoints": [
|
||||
{
|
||||
"name": "default",
|
||||
"subject": "svc.echo",
|
||||
"queue_group": "q",
|
||||
"num_requests": 0,
|
||||
"num_errors": 0,
|
||||
"last_error": "",
|
||||
"processing_time": 0,
|
||||
"average_processing_time": 0
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
## Examples
|
||||
|
||||
For more detailed examples, refer to the `./test/example_test.go` directory in
|
||||
this package.
|
||||
|
||||
## Documentation
|
||||
|
||||
The complete documentation is available on
|
||||
[GoDoc](https://godoc.org/github.com/nats-io/nats.go/micro).
|
||||
+211
@@ -0,0 +1,211 @@
|
||||
// Copyright 2022-2023 The NATS Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package micro
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
type (
|
||||
// Handler is used to respond to service requests.
|
||||
Handler interface {
|
||||
Handle(Request)
|
||||
}
|
||||
|
||||
// HandlerFunc is a function implementing [Handler].
|
||||
// It allows using a function as a request handler, without having to implement Handle
|
||||
// on a separate type.
|
||||
HandlerFunc func(Request)
|
||||
|
||||
// Request represents service request available in the service handler.
|
||||
// It exposes methods to respond to the request, as well as
|
||||
// getting the request data and headers.
|
||||
Request interface {
|
||||
// Respond sends the response for the request.
|
||||
// Additional headers can be passed using [WithHeaders] option.
|
||||
Respond([]byte, ...RespondOpt) error
|
||||
|
||||
// RespondJSON marshals the given response value and responds to the request.
|
||||
// Additional headers can be passed using [WithHeaders] option.
|
||||
RespondJSON(any, ...RespondOpt) error
|
||||
|
||||
// Error prepares and publishes error response from a handler.
|
||||
// A response error should be set containing an error code and description.
|
||||
// Optionally, data can be set as response payload.
|
||||
Error(code, description string, data []byte, opts ...RespondOpt) error
|
||||
|
||||
// Data returns request data.
|
||||
Data() []byte
|
||||
|
||||
// Headers returns request headers.
|
||||
Headers() Headers
|
||||
|
||||
// Subject returns underlying NATS message subject.
|
||||
Subject() string
|
||||
|
||||
// Reply returns underlying NATS message reply subject.
|
||||
Reply() string
|
||||
}
|
||||
|
||||
// Headers is a wrapper around [*nats.Header]
|
||||
Headers nats.Header
|
||||
|
||||
// RespondOpt is a function used to configure [Request.Respond] and [Request.RespondJSON] methods.
|
||||
RespondOpt func(*nats.Msg)
|
||||
|
||||
// request is a default implementation of Request interface
|
||||
request struct {
|
||||
msg *nats.Msg
|
||||
respondError error
|
||||
}
|
||||
|
||||
serviceError struct {
|
||||
Code string `json:"code"`
|
||||
Description string `json:"description"`
|
||||
}
|
||||
)
|
||||
|
||||
var (
|
||||
ErrRespond = errors.New("NATS error when sending response")
|
||||
ErrMarshalResponse = errors.New("marshaling response")
|
||||
ErrArgRequired = errors.New("argument required")
|
||||
)
|
||||
|
||||
func (fn HandlerFunc) Handle(req Request) {
|
||||
fn(req)
|
||||
}
|
||||
|
||||
// ContextHandler is a helper function used to utilize [context.Context]
|
||||
// in request handlers.
|
||||
func ContextHandler(ctx context.Context, handler func(context.Context, Request)) Handler {
|
||||
return HandlerFunc(func(req Request) {
|
||||
handler(ctx, req)
|
||||
})
|
||||
}
|
||||
|
||||
// Respond sends the response for the request.
|
||||
// Additional headers can be passed using [WithHeaders] option.
|
||||
func (r *request) Respond(response []byte, opts ...RespondOpt) error {
|
||||
respMsg := &nats.Msg{
|
||||
Data: response,
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(respMsg)
|
||||
}
|
||||
|
||||
if err := r.msg.RespondMsg(respMsg); err != nil {
|
||||
r.respondError = fmt.Errorf("%w: %s", ErrRespond, err)
|
||||
return r.respondError
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RespondJSON marshals the given response value and responds to the request.
|
||||
// Additional headers can be passed using [WithHeaders] option.
|
||||
func (r *request) RespondJSON(response any, opts ...RespondOpt) error {
|
||||
resp, err := json.Marshal(response)
|
||||
if err != nil {
|
||||
return ErrMarshalResponse
|
||||
}
|
||||
return r.Respond(resp, opts...)
|
||||
}
|
||||
|
||||
// Error prepares and publishes error response from a handler.
|
||||
// A response error should be set containing an error code and description.
|
||||
// Optionally, data can be set as response payload.
|
||||
func (r *request) Error(code, description string, data []byte, opts ...RespondOpt) error {
|
||||
if code == "" {
|
||||
return fmt.Errorf("%w: error code", ErrArgRequired)
|
||||
}
|
||||
if description == "" {
|
||||
return fmt.Errorf("%w: description", ErrArgRequired)
|
||||
}
|
||||
response := &nats.Msg{
|
||||
Header: nats.Header{
|
||||
ErrorHeader: []string{description},
|
||||
ErrorCodeHeader: []string{code},
|
||||
},
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(response)
|
||||
}
|
||||
|
||||
response.Data = data
|
||||
if err := r.msg.RespondMsg(response); err != nil {
|
||||
r.respondError = err
|
||||
return err
|
||||
}
|
||||
r.respondError = &serviceError{
|
||||
Code: code,
|
||||
Description: description,
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// WithHeaders can be used to configure response with custom headers.
|
||||
func WithHeaders(headers Headers) RespondOpt {
|
||||
return func(m *nats.Msg) {
|
||||
if m.Header == nil {
|
||||
m.Header = nats.Header(headers)
|
||||
return
|
||||
}
|
||||
|
||||
for k, v := range headers {
|
||||
m.Header[k] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Data returns request data.
|
||||
func (r *request) Data() []byte {
|
||||
return r.msg.Data
|
||||
}
|
||||
|
||||
// Headers returns request headers.
|
||||
func (r *request) Headers() Headers {
|
||||
return Headers(r.msg.Header)
|
||||
}
|
||||
|
||||
// Subject returns underlying NATS message subject.
|
||||
func (r *request) Subject() string {
|
||||
return r.msg.Subject
|
||||
}
|
||||
|
||||
// Reply returns underlying NATS message reply subject.
|
||||
func (r *request) Reply() string {
|
||||
return r.msg.Reply
|
||||
}
|
||||
|
||||
// Get gets the first value associated with the given key.
|
||||
// It is case-sensitive.
|
||||
func (h Headers) Get(key string) string {
|
||||
return nats.Header(h).Get(key)
|
||||
}
|
||||
|
||||
// Values returns all values associated with the given key.
|
||||
// It is case-sensitive.
|
||||
func (h Headers) Values(key string) []string {
|
||||
return nats.Header(h).Values(key)
|
||||
}
|
||||
|
||||
func (e *serviceError) Error() string {
|
||||
return fmt.Sprintf("%s:%s", e.Code, e.Description)
|
||||
}
|
||||
+1023
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user