Noțiuni de bază

Despre NATS

Aplicațiile și serviciile software trebuie să schimbe date. NATS este o infrastructură care permite acest schimb de date, segmentat sub formă de mesaje. Numim acest lucru "message oriented middleware".
Cu NATS, dezvoltatorii de aplicații pot:
Să construiască aplicații client-server distribuite și scalabile fără efort.
Să stocheze și să distribuie date în timp real, într-un mod general. Acest lucru poate fi realizat flexibil în diverse medii, limbaje, furnizori de cloud și sisteme on-premises.
What is NATS, NATS docs

NATS 是一个 Go 语言编写的消息代理。

NATS încorporat

Dacă aplicația dumneavoastră este scrisă în Go și dacă se potrivește cazului de utilizare și scenariilor de implementare, puteți chiar să încorporați un server NATS în interiorul aplicației.

Embedding NATS, NATS docs

Și există o particularitate a NATS: pentru aplicațiile construite în Go, este suportat modul embedded.
Cu alte cuvinte, în loc de structura obișnuită a unui broker de mesaje, în care un server de broker separat este pornit și comunicarea se realizează prin clienții aplicației cu acel server, brokerul propriu-zis poate fi încorporat (embed) în aplicația construită în Go.

Beneficiile și cazurile de utilizare ale NATS încorporat

Există un videoclip YouTube bine explicat, astfel că il substituim cu linkul videoclipului.
Chiar și fără implementarea unui server de broker de mesaje separat, se poate crea o aplicație modular monolith și se pot realiza separate of concern, valorificând avantajul încorporării NATS ca embedded. În plus, devine posibilă și implementarea sub forma unui single binary.
Pe lângă platformele fără rețea (wasm), poate fi utilizat cu folos și în aplicații offline-first.

Exemplu din documentația oficială

 1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6
 7    "github.com/nats-io/nats-server/v2/server"
 8    "github.com/nats-io/nats.go"
 9)
10
11func main() {
12    opts := &server.Options{}
13
14    // Initialize new server with options
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Start the server via goroutine
22    go ns.Start()
23
24    // Wait for server to be ready for connections
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Connect to server
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Subscribe to the subject
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Print message data
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Shutdown the server (optional)
45        ns.Shutdown()
46    })
47
48    // Publish data to the subject
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Wait for server shutdown
52    ns.WaitForShutdown()
53}

Acesta este exemplul Embedded NATS prezentat în documentația oficială NATS, însă, dacă se urmează acest cod de exemplu, comunicarea nu se realizează în modul embedding.

1Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:34:20
2                                                                     in 0.017s (0)
3...
4tcp4       0      0  127.0.0.1.4222         127.0.0.1.63769        TIME_WAIT

Prin comanda watch 'netstat -an | grep 127.0.0.1', care verifică rețeaua care circulă către localhost (127.0.0.1), și prin rularea fișierului Go cu go run ., se poate observa că apar noi cereri de rețea care pornesc de la portul implicit al NATS, 4222.

Configurațiile corecte pentru modul embedding

Pentru a comunica în modul embedded așa cum se dorește, sunt necesare două opțiuni:
- Client: trebuie transmisă opțiunea InProcessServer.
- Server: în Server.Options trebuie specificat explicit flag-ul DontListen cu valoarea true.
Aceste aspecte nu au fost documentate oficial, iar originea acestei funcționalități poate fi urmărită în acest PR.
Acest PR adaugă trei lucruri:
1. Funcția InProcessConn() pentru Server, care construiește un net.Pipe pentru a obține o conexiune la serverul NATS fără a utiliza socket-uri TCP
2. Opțiunea DontListen, care îi spune serverului NATS să nu asculte pe listener-ul TCP obișnuit
3. Canalul startupComplete, care este închis chiar înainte de a începe AcceptLoop, iar readyForConnections va aștepta după el
Motivația principală este că avem o aplicație care poate rula fie în modul monolith (single-process), fie în modul polylith (multi-process). Dorim să putem folosi NATS în ambele moduri, pentru simplitate, însă modul monolith trebuie să poată deservi o varietate de platforme unde deschiderea conexiunilor prin socket nu are sens (mobile) sau pur și simplu nu este posibilă (WASM). Aceste modificări ne vor permite să utilizăm NATS integral in-process.
Un PR însoțitor nats-io/nats.go#774 adaugă suport pe partea de client.
Acesta este primul meu PR la acest proiect, așa că îmi cer scuze din timp dacă am omis ceva evident.
/cc @nats-io/core

Exemplu funcțional pentru modul embedded

 1package main
 2
 3import (
 4	"fmt"
 5	"time"
 6
 7	"github.com/nats-io/nats-server/v2/server"
 8	"github.com/nats-io/nats.go"
 9)
10
11func main() {
12	opts := &server.Options{
13		// for configuring the embeded NATS server
14		// set DonListen as true
15		DontListen: true,
16	}
17
18	// Initialize new server with options
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Start the server via goroutine
26	go ns.Start()
27
28	// Wait for server to be ready for connections
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Connect to server via in-process connection
34	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
35
36	if err != nil {
37		panic(err)
38	}
39
40	subject := "my-subject"
41
42	// Subscribe to the subject
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Print message data
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Shutdown the server (optional)
49		ns.Shutdown()
50	})
51
52	// Publish data to the subject
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Wait for server shutdown
56	ns.WaitForShutdown()
57}
58

1Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:37:50
2                                                                     in 0.023s (0)
3...no additional logs
4

Acum se poate observa că nu mai apare niciun salt de rețea suplimentar, cum era intenționat.

În profunzime

TL;DR

diagram1

Aceasta este diagrama de tip sequence care arată ce funcții sunt apelate intern și cum funcționează atunci când acest cod este executat din main.go. Esența este următoarea:
- Prin DontListen: true, serverul omite faza de ascultare a clienților numită AcceptLoop.
- Dacă opțiunea InProcessServer a Connect-ului clientului este activată, se creează o conexiune in-memory, se construiește un pipe prin net.Pipe, iar capătul pipe-ului este returnat clientului ca tip net.Conn.
- Clientul și serverul realizează comunicarea in-process prin această conexiune.

Server

AcceptLoop

1// nats-server/server/server.go
2
3// Wait for clients.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}

Mai întâi, dacă DontListen este true, se omite faza de ascultare a clienților numită AcceptLoop.

 1// nats-server/server/server.go
 2
 3// AcceptLoop is exported for easier testing.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// If we were to exit before the listener is setup properly,
 6	// make sure we close the channel.
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// Snapshot server options.
18	opts := s.getOpts()
19
20	// Setup state that can enable shutdown
21	s.mu.Lock()
22	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
23	l, e := natsListen("tcp", hp)
24	s.listenerErr = e
25	if e != nil {
26		s.mu.Unlock()
27		s.Fatalf("Error listening on port: %s, %q", hp, e)
28		return
29	}
30	s.Noticef("Listening for client connections on %s",
31		net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))
32
33	// Alert of TLS enabled.
34	if opts.TLSConfig != nil {
35		s.Noticef("TLS required for client connections")
36		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
37			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
38		}
39	}
40
41	// If server was started with RANDOM_PORT (-1), opts.Port would be equal
42	// to 0 at the beginning this function. So we need to get the actual port
43	if opts.Port == 0 {
44		// Write resolved port back to options.
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// Now that port has been set (if it was set to RANDOM), set the
49	// server's info Host/Port with either values from Options or
50	// ClientAdvertise.
51	if err := s.setInfoHostPort(); err != nil {
52		s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
53		l.Close()
54		s.mu.Unlock()
55		return
56	}
57	// Keep track of client connect URLs. We may need them later.
58	s.clientConnectURLs = s.getClientConnectURLs()
59	s.listener = l
60
61	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
62		func(_ error) bool {
63			if s.isLameDuckMode() {
64				// Signal that we are not accepting new clients
65				s.ldmCh <- true
66				// Now wait for the Shutdown...
67				<-s.quitCh
68				return true
69			}
70			return false
71		})
72	s.mu.Unlock()
73
74	// Let the caller know that we are ready
75	close(clr)
76	clr = nil
77}

Ca observație, funcția AcceptLoop parcurge următorii pași. Părțile legate de comunicarea în rețea, cum sunt TLS sau hostPort, nu sunt necesare în cazul comunicării in-process, așadar pot fi omise fără probleme.

Client

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect will attempt to connect to the NATS system.
 5// The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222
 6// Comma separated arrays are also supported, e.g. urlA, urlB.
 7// Options start with the defaults but can be overridden.
 8// To connect to a NATS Server's websocket port, use the `ws` or `wss` scheme, such as
 9// `ws://localhost:8080`. Note that websocket schemes cannot be mixed with others (nats/tls).
10func Connect(url string, options ...Option) (*Conn, error) {
11	opts := GetDefaultOptions()
12	opts.Servers = processUrlString(url)
13	for _, opt := range options {
14		if opt != nil {
15			if err := opt(&opts); err != nil {
16				return nil, err
17			}
18		}
19	}
20	return opts.Connect()
21}

 1// nats-go/nats.go
 2
 3// Options can be used to create a customized connection.
 4type Options struct {
 5	// Url represents a single NATS server url to which the client
 6	// will be connecting. If the Servers option is also set, it
 7	// then becomes the first server in the Servers array.
 8	Url string
 9
10	// InProcessServer represents a NATS server running within the
11	// same process. If this is set then we will attempt to connect
12	// to the server directly rather than using external TCP conns.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}

Funcția Connect, care realizează conectarea dintre serverul NATS și clientul NATS, primește URL-ul clientului și opțiunile de conectare, iar structura Options, în care sunt adunate aceste opțiuni, conține un câmp numit InProcessServer de tipul interfață InProcesConnProvider.

1// main.go of example code
2
3// Initialize new server with options
4ns, err := server.NewServer(opts)
5
6//...
7
8// Connect to server via in-process connection
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

Când clientul NATS apelează Connect și transmite nats.InProcessServer(ns) ca valoare a câmpului InProcessServer,

 1// nats-go/nats.go
 2
 3// InProcessServer is an Option that will try to establish a direction to a NATS server
 4// running within the process instead of dialing via TCP.
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}

opțiunea InProcessServer este înlocuită cu serverul NATS încorporat,

 1// nats-go/nats.go
 2
 3// createConn will connect to the server and wrap the appropriate
 4// bufio structures. It will do the right thing when an existing
 5// connection is in place.
 6func (nc *Conn) createConn() (err error) {
 7	if nc.Opts.Timeout < 0 {
 8		return ErrBadTimeout
 9	}
10	if _, cur := nc.currentServer(); cur == nil {
11		return ErrNoServers
12	}
13
14	// If we have a reference to an in-process server then establish a
15	// connection using that.
16	if nc.Opts.InProcessServer != nil {
17		conn, err := nc.Opts.InProcessServer.InProcessConn()
18		if err != nil {
19			return fmt.Errorf("failed to get in-process connection: %w", err)
20		}
21		nc.conn = conn
22		nc.bindToNewConn()
23		return nil
24	}
25
26	//...
27}

Această interfață este utilizată în funcția createConn, care creează conexiunea. Dacă opțiunea InProcessServer nu este nil (adică este validă), se execută InProcessConn al lui InProcessServer din opțiune,

 1// nats-server/server/server.go
 2
 3// InProcessConn returns an in-process connection to the server,
 4// avoiding the need to use a TCP listener for local connectivity
 5// within the same process. This can be used regardless of the
 6// state of the DontListen option.
 7func (s *Server) InProcessConn() (net.Conn, error) {
 8	pl, pr := net.Pipe()
 9	if !s.startGoRoutine(func() {
10		s.createClientInProcess(pl)
11		s.grWG.Done()
12	}) {
13		pl.Close()
14		pr.Close()
15		return nil, fmt.Errorf("failed to create connection")
16	}
17	return pr, nil
18}

Se apelează InProcessConn implementată în server.
Această funcție este apelată din nats.go, clientul Go pentru NATS, atunci când InProcessServer al nc (conexiunii NATS) nu este nil; ea creează conexiunea (net.Conn), care este apoi legată de conexiunea serverului.

Interfață definită de consumator în Go

Un tip implementează o interfață prin simpla implementare a metodelor sale. Nu există o declarație explicită a intenției, nu există cuvântul cheie "implements". Interfețele implicite decuplează definiția unei interfețe de implementarea sa, care poate apărea apoi în orice pachet, fără aranjamente prealabile.

Interfaces are implemented implicitly, A Tour of Go

Dacă un tip există doar pentru a implementa o interfață și nu va avea niciodată metode exportate dincolo de acea interfață, nu este necesar să exportăm tipul în sine.

Generality, Effective Go

Această abordare a interfețelor reflectă bine ceea ce în Go se numește interfață definită de consumator și tipizare structurală (duck typing), astfel că merită să prezentăm și acest subiect.

 1// nats-go/nats.go
 2
 3// Options can be used to create a customized connection.
 4type Options struct {
 5	// Url represents a single NATS server url to which the client
 6	// will be connecting. If the Servers option is also set, it
 7	// then becomes the first server in the Servers array.
 8	Url string
 9
10	// InProcessServer represents a NATS server running within the
11	// same process. If this is set then we will attempt to connect
12	// to the server directly rather than using external TCP conns.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}

Revenind la cod: în clientul nats.go, câmpul de structură al opțiunii InProcessServer este definit ca interfața InProcessConnProvider, care realizează doar InProcessConn.

 1// nats-server/server/server.go
 2
 3// InProcessConn returns an in-process connection to the server,
 4// avoiding the need to use a TCP listener for local connectivity
 5// within the same process. This can be used regardless of the
 6// state of the DontListen option.
 7func (s *Server) InProcessConn() (net.Conn, error) {
 8	pl, pr := net.Pipe()
 9	if !s.startGoRoutine(func() {
10		s.createClientInProcess(pl)
11		s.grWG.Done()
12	}) {
13		pl.Close()
14		pr.Close()
15		return nil, fmt.Errorf("failed to create connection")
16	}
17	return pr, nil
18}

Totuși, tipul care este transmis acolo este Server din nats-server, care îndeplinește nu doar InProcessConn, ci și multe alte funcții.
Acest lucru se explică prin faptul că, în acest context, interesul clientului este doar dacă a fost furnizată interfața InProcessConn sau nu, iar celelalte aspecte nu sunt deosebit de relevante.
Prin urmare, nats.go client folosește doar interfața InProcessConnProvider, o interfață definită de consumator care specifică doar funcționalitatea InProcessConn() (net.Conn, error).

Concluzie

Am tratat pe scurt modul embedded al NATS, modul său de funcționare și interfața definită de consumator din Go, care poate fi observată în codul NATS.
Sper că aceste informații vor fi utile celor care utilizează NATS în scopurile menționate mai sus, și închei aici acest articol.