Aloitus

Tietoa NATS:sta

Ohjelmistosovellukset ja -palvelut tarvitsevat tiedon vaihtoa. NATS on infrastruktuuri, joka mahdollistaa tällaisen tiedonvaihdon viestien muodossa. Kutsumme tätä "sanomavälitteiseksi väliohjelmistoksi" (message oriented middleware).
NATS:n avulla sovelluskehittäjät voivat:
Rakentaa hajautettuja ja skaalautuvia asiakas-palvelin-sovelluksia vaivattomasti.
Tallentaa ja jakaa dataa reaaliajassa yleisellä tavalla. Tämä voidaan joustavasti toteuttaa eri ympäristöissä, ohjelmointikielillä, pilvipalveluntarjoajilla ja omissa järjestelmissä.
What is NATS, NATS docs

NATS on Go-kielellä toteutettu viestinvälittäjä (message broker).

Upotettu NATS

Jos sovelluksesi on toteutettu Go-kielellä ja se sopii käyttötapaukseesi ja käyttöönottoskenaarioihisi, voit jopa upottaa NATS-palvelimen sovellukseesi.

Embedding NATS, NATS docs

Lisäksi NATS:ssa on erityispiirre: Go-kielellä toteutettuja sovelluksia varten tuetaan upotettua tilaa (embedded mode).
Toisin sanoen sen sijaan, että viestinvälittäjä käynnistettäisiin erillisenä palvelimena ja sovelluksen asiakkaat kommunikoisivat sen kautta, kuten viestinvälittäjillä yleensä tehdään, viestinvälittäjän itsensä voi upottaa (embed) Go-kielellä kirjoitettuun sovellukseen.

Upotetun NATS:n hyödyt ja käyttötapaukset

Aiheesta on hyvin selittävä YouTube-video, joten käytän videon linkkiä sen sijaan, että selittäisin asian itse.
Vaikka erillistä viestinvälittäjäpalvelinta ei otettaisi käyttöön, voidaan rakentaa modular monolith -sovellus ja saavuttaa separation of concerns, ja samalla hyödyntää sitä, että NATS voidaan upottaa sovellukseen. Lisäksi mahdollistuu single binary -käyttöönotto.
Sen lisäksi, että se soveltuu alustoihin ilman verkkoyhteyttä (wasm), sitä voi hyödyntää tehokkaasti myös offline-first-sovelluksissa.

Esimerkki virallisessa dokumentaatiossa

 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    // Alusta uusi palvelin asetuksilla
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Käynnistä palvelin goroutinena
22    go ns.Start()
23
24    // Odota, että palvelin on valmis vastaanottamaan yhteyksiä
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Yhdistä palvelimeen
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Tilaa aihe
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Tulosta viestin data
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Sammuta palvelin (valinnainen)
45        ns.Shutdown()
46    })
47
48    // Julkaise data aiheeseen
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Odota palvelimen sammumista
52    ns.WaitForShutdown()
53}

Tämä on NATS:n virallisen dokumentaation linkittämä esimerkki upotetusta NATS:sta, mutta jos tätä esimerkkikoodia seurataan sellaisenaan, kommunikaatio ei tapahdu upotetussa tilassa (embedding mode).

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

Kun seurataan localhostin (127.0.0.1) verkkoliikennettä komennolla watch 'netstat -an | grep 127.0.0.1' ja suoritetaan Go-tiedosto komennolla go run ., nähdään, että NATS:n oletusportista 4222 lähtee uusia verkkopyyntöjä.

Oikeat asetukset upotetulle tilalle

Jotta kommunikaatio tapahtuisi halutulla tavalla upotetussa tilassa, tarvitaan seuraavat kaksi asetusta.
- Asiakas: InProcessServer-asetus on annettava.
- Palvelin: Server.Options-rakenteessa on asetettava lippu DontListen arvoon true.
Näitä ominaisuuksia ei ole virallisesti dokumentoitu, ja tämän ominaisuuden alkuperän voi selvittää tästä PR:stä.
This PR adds three things:
1. InProcessConn() function to Server which builds a net.Pipe to get a connection to the NATS server without using TCP sockets
2. DontListen option which tells the NATS server not to listen on the usual TCP listener
3. startupComplete channel, which is closed right before we start AcceptLoop, and readyForConnections will wait for it
The main motivation for this is that we have an application that can run either in a monolith (single-process) mode or a polylith (multi-process) mode. We'd like to be able to use NATS for both modes for simplicity, but the monolith mode has to be able to cater for a variety of platforms where opening socket connections either doesn't make sense (mobile) or just isn't possible (WASM). These changes will allow us to use NATS entirely in-process instead.
An accompanying PR nats-io/nats.go#774 adds support to the client side.
This is my first PR to this project so apologies in advance if I've missed anything obvious anywhere.
/cc @nats-io/core

Tämä PR lisää kolme asiaa:
InProcessConn()-funktio Server-tyyppiin, joka muodostaa net.Pipe-yhteyden NATS-palvelimeen ilman TCP-socketeja
DontListen-asetus, joka kertoo NATS-palvelimelle, ettei sen tule kuunnella tavanomaista TCP-kuuntelijaa
startupComplete-kanava, joka suljetaan juuri ennen AcceptLoop-silmukan käynnistämistä, ja readyForConnections odottaa sitä
Tämän pääasiallinen motivaatio on, että meillä on sovellus, joka voi toimia joko monoliittisessa (yksiprosessisessa) tilassa tai polyliittisessa (moniprosessisessa) tilassa. Haluaisimme voida käyttää NATS:ia molemmissa tiloissa yksinkertaisuuden vuoksi, mutta monoliittisen tilan on pystyttävä palvelemaan useita alustoja, joissa socket-yhteyksien avaaminen joko ei ole järkevää (mobiili) tai ei ole mahdollista lainkaan (WASM). Nämä muutokset mahdollistavat NATS:n käyttämisen kokonaan prosessin sisällä.
Asiaan liittyvä PR nats-io/nats.go#774 lisää tuen asiakaspuolelle.
Tämä on ensimmäinen PR tähän projektiin, joten pahoittelen etukäteen, jos olen jättänyt jotain ilmeistä huomiotta.
/cc @nats-io/core

Toimiva esimerkki upotetulle tilalle

 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		// Upotetun NATS-palvelimen asetukset
14		// aseta DonListen arvoon true
15		DontListen: true,
16	}
17
18	// Alusta uusi palvelin asetuksilla
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Käynnistä palvelin goroutinena
26	go ns.Start()
27
28	// Odota, että palvelin on valmis vastaanottamaan yhteyksiä
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Yhdistä palvelimeen prosessinsisäisen yhteyden kautta
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	// Tilaa aihe
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Tulosta viestin data
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Sammuta palvelin (valinnainen)
49		ns.Shutdown()
50	})
51
52	// Julkaise data aiheeseen
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Odota palvelimen sammumista
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...ei lisälokeja
4

Nyt nähdään, että ylimääräisiä verkkohyppyjä ei synny, kuten oli tarkoitus.

Konepellin alla

TL;DR

diagram1

Kun tämä koodi suoritetaan tiedostosta main.go, kyseessä on sekvenssikaavio, joka kuvaa, mitä funktioita suoritetaan sisäisesti ja missä järjestyksessä. Pääpiirteissään se on seuraava.
- DontListen: true -asetuksen ansiosta palvelin ohittaa AcceptLoop-vaiheen, eli asiakkaiden kuunteluvaiheen.
- Jos asiakkaan Connect-asetuksista InProcessServer on aktivoitu, luodaan muistinvarainen yhteys: rakennetaan net.Pipe-putki ja palautetaan putken toinen pää asiakkaalle net.Conn-tyyppinä.
- Asiakas ja palvelin kommunikoivat tämän yhteyden kautta prosessinsisäisesti.

Palvelin

AcceptLoop

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

Ensin, jos DontListen on true, palvelin ohittaa AcceptLoop-vaiheen, eli asiakkaiden kuunteluvaiheen.

 1// nats-server/server/server.go
 2
 3// AcceptLoop on exportattu helpompaa testausta varten.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// Jos poistuisimme ennen kuli kuuntelija on alustettu kunnolla,
 6	// varmista, että suljemme kanavan.
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// Ota palvelimen asetuksista kopio.
18	opts := s.getOpts()
19
20	// Alusta tila, joka mahdollistaa sammutuksen
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	// Ilmoita, että TLS on käytössä.
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	// Jos palvelin käynnistettiin RANDOM_PORT (-1) -asetuksella, opts.Port on
42	// tämän funktion alussa 0. Siksi tarvitsemme todellisen portin.
43	if opts.Port == 0 {
44		// Kirjoita selvitetty portti takaisin asetuksiin.
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// Nyt kun portti on asetettu (jos se oli RANDOM), aseta
49	// palvelimen info Host/Port joko asetusten tai
50	// ClientAdvertise-arvon mukaan.
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	// Pidä kirjaa asiakkaiden yhdistys-URL-osoitteista. Saatamme tarvita niitä myöhemmin.
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				// Ilmoita, ettei uusia asiakkaita hyväksytä
65				s.ldmCh <- true
66				// Odota nyt sammutusta...
67				<-s.quitCh
68				return true
69			}
70			return false
71		})
72	s.mu.Unlock()
73
74	// Ilmoita kutsujalle, että olemme valmiita
75	close(clr)
76	clr = nil
77}

Huomautuksena: AcceptLoop-funktio suorittaa seuraavat vaiheet. TLS ja hostPort liittyvät verkkoliikenteeseen, eikä niitä tarvita prosessinsisäisessä kommunikaatiossa, joten ne voidaan jättää pois.

Asiakas

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect yrittää muodostaa yhteyden NATS-järjestelmään.
 5// URL voi sisältää käyttäjätunnuksen ja salasanan, esim. nats://derek:pass@localhost:4222.
 6// Pilkulla erotettuja taulukoita tuetaan myös, esim. urlA, urlB.
 7// Asetukset alkavat oletusarvoista, mutta niitä voidaan muuttaa.
 8// Yhdistääksesi NATS-palvelimen websocket-porttiin, käytä skeemaa `ws` tai `wss`, esim.
 9// `ws://localhost:8080`. Huomaa, että websocket-skeemoja ei voi yhdistää muihin (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-tyyppiä voidaan käyttää mukautetun yhteyden luomiseen.
 4type Options struct {
 5	// Url edustaa yhtä NATS-palvelimen URL-osoitetta, johon asiakas
 6	// yhdistää. Jos myös Servers-asetus on annettu, siitä tulee
 7	// Servers-taulukon ensimmäinen palvelin.
 8	Url string
 9
10	// InProcessServer edustaa samassa prosessissa käynnissä olevaa
11	// NATS-palvelinta. Jos tämä on asetettu, yritämme yhdistää
12	// palvelimeen suoraan sen sijaan, että käyttäisimme ulkoisia TCP-yhteyksiä.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

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

Connect-funktio, joka yhdistää NATS-palvelimen ja NATS-asiakkaan, ottaa vastaan client-URL-osoitteen ja connect-asetuksia. Asetukset kokoava Options-rakenne sisältää kentän InProcessServer, jonka tyyppi on InProcessConnProvider-rajapinta.

1// main.go of example code
2
3// Alusta uusi palvelin asetuksilla
4ns, err := server.NewServer(opts)
5
6//...
7
8// Yhdistä palvelimeen prosessinsisäisen yhteyden kautta
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

Kun NATS-asiakas suorittaa Connect-kutsun ja välittää InProcessServer-kentälle arvon nats.InProcessServer(ns),

 1// nats-go/nats.go
 2
 3// InProcessServer on Option, joka yrittää muodostaa yhteyden samassa
 4// prosessissa käynnissä olevaan NATS-palvelimeen TCP-yhteyden sijaan.
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}

asetuksen InProcessServer-kenttä korvataan upotetulla NATS-palvelimella,

 1// nats-go/nats.go
 2
 3// createConn muodostaa yhteyden palvelimeen ja kääröi sen sopiviin
 4// bufio-rakenteisiin. Se toimii oikein, kun olemassa oleva
 5// yhteys on jo käytössä.
 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	// Jos meillä on viite prosessinsisäiseen palvelimeen, muodosta
15	// yhteys sen kautta.
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}

Tämä rajapinta toimii yhteyden luovassa createConn-funktiossa. Jos InProcessServer-asetus ei ole nil (eli on kelvollinen), kutsutaan asetuksen InProcessServer-kentän InProcessConn-metodia,

 1// nats-server/server/server.go
 2
 3// InProcessConn palauttaa prosessinsisäisen yhteyden palvelimeen
 4// ilman tarvetta TCP-kuuntelijalle paikallista yhteyttä varten
 5// samassa prosessissa. Tätä voidaan käyttää riippumatta
 6// DontListen-asetuksen tilasta.
 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}

joka kutsuu palvelimelle toteutettua InProcessConn-metodia.
Tämä funktio kutsutaan nats.go-asiakkaassa, kun nc:n (NATS-yhteyden) InProcessServer ei ole nil. Se luo yhteyden (net.Conn) ja sitoo sen palvelimen yhteyteen.

Go-kielen kuluttajalähtöinen rajapinta

A type implements an interface by implementing its methods. There is no explicit declaration of intent, no "implements" keyword. Implicit interfaces decouple the definition of an interface from its implementation, which could then appear in any package without prearrangement.

Tyyppi toteuttaa rajapinnan toteuttamalla sen metodit. Aikomusta ei ilmoiteta erikseen, eikä ole "implements"-avainsanaa. Implisiittiset rajapinnat erottavat rajapinnan määrittelyn sen toteutuksesta, jolloin rajapinta voi esiintyä missä tahansa paketissa ilman ennakkosopimusta.

Interfaces are implemented implicitly, A Tour of Go

If a type exists only to implement an interface and will never have exported methods beyond that interface, there is no need to export the type itself.

Jos tyyppi on olemassa vain rajapinnan toteuttamista varten eikä sillä koskaan ole exportattuja metodeja tuon rajapinnan lisäksi, itse tyyppiä ei tarvitse exportata.

Generality, Effective Go

Tämä rajapintasuunnittelu ilmentää hyvin Go-kielen usein mainitsemaa kuluttajan määrittelemää rajapintaa (consumer defined interface) ja rakenteellista tyypitystä (structural typing, eli duck typing), joten haluan esitellä myös tämän aiheen.

 1// nats-go/nats.go
 2
 3// Options-tyyppiä voidaan käyttää mukautetun yhteyden luomiseen.
 4type Options struct {
 5	// Url edustaa yhtä NATS-palvelimen URL-osoitetta, johon asiakas
 6	// yhdistää. Jos myös Servers-asetus on annettu, siitä tulee
 7	// Servers-taulukon ensimmäinen palvelin.
 8	Url string
 9
10	// InProcessServer edustaa samassa prosessissa käynnissä olevaa
11	// NATS-palvelinta. Jos tämä on asetettu, yritämme yhdistää
12	// palvelimeen suoraan sen sijaan, että käyttäisimme ulkoisia TCP-yhteyksiä.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

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

Palataan vielä koodiin. NATS.go-asiakkaassa InProcessServer-asetuksen kenttä on määritelty InProcessConnProvider-rajapinnaksi, joka suorittaa vain InProcessConn-metodin.

 1// nats-server/server/server.go
 2
 3// InProcessConn palauttaa prosessinsisäisen yhteyden palvelimeen
 4// ilman tarvetta TCP-kuuntelijalle paikallista yhteyttä varten
 5// samassa prosessissa. Tätä voidaan käyttää riippumatta
 6// DontListen-asetuksen tilasta.
 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}

Sen tyyppi on kuitenkin nats-serverin Server, joka suorittaa paljon muitakin toimintoja kuin pelkän InProcessConn-metodin.
Tämä johtuu siitä, että asiakkaan kannalta olennaista on vain se, tarjoaako palvelin InProcessConn-rajapinnan, eikä muut ominaisuudet ole merkityksellisiä.
Siksi nats.go-asiakas käyttää vain InProcessConnProvider-rajapintaa, joka määrittelee pelkän InProcessConn() (net.Conn, error) -toiminnon. Se on kuluttajan määrittelemä rajapinta (consumer defined interface).

Yhteenveto

Käsittelimme lyhyesti NATS:n upotettua tilaa ja sen toimintaperiaatetta sekä Go-kielen kuluttajan määrittelemää rajapintaa, joka näkyy NATS:n koodissa.
Toivon, että tästä tiedosta on hyötyä niille, jotka käyttävät NATS:ia samoihin tarkoituksiin, ja päätän kirjoituksen tähän.