Pierwsze kroki

O NATS

Aplikacje programowe i usługi muszą wymieniać dane. NATS to infrastruktura umożliwiająca taką wymianę danych, podzieloną na postaci wiadomości. Nazywamy ją "message oriented middleware" (middleware zorientowane na wiadomości).

Dzięki NATS programiści aplikacji mogą:

• Bez wysiłku tworzyć rozproszone i skalowalne aplikacje typu klient-serwer.
• Przechowywać i rozprowadzać dane w czasie rzeczywistym w sposób ogólny. Można to elastycznie osiągnąć w różnych środowiskach, językach, u dostawców chmury oraz w systemach lokalnych (on-premises).

What is NATS, NATS docs

• NATS 是 broker wiadomości napisany w Go.

Osadzony NATS (Embedded NATS)

Jeśli Twoja aplikacja jest napisana w Go i jeśli odpowiada to Twojemu przypadkowi użycia oraz scenariuszom wdrożenia, możesz nawet osadzić serwer NATS wewnątrz swojej aplikacji.

Embedding NATS, NATS docs

• Istnieje jednak pewna szczególna cecha NATS: w przypadku aplikacji napisanych w Go obsługiwany jest tryb embedded.
• Oznacza to, że zamiast powszechnego dla brokerów wiadomości modelu, w którym uruchamia się osobny serwer brokera i komunikuje się z nim klient aplikacji, sam broker można osadzić (embed) wewnątrz aplikacji napisanej w Go.

Korzyści i przypadki użycia osadzonego NATS

• Istnieje dobrze wyjaśniający to zagadnienie film na YouTube, dlatego odsyłam do linku z filmem.
• Nawet bez wdrażania oddzielnego serwera brokera wiadomości można tworzyć modular monolith application i osiągać separate of concern, czerpiąc korzyść z możliwości osadzenia NATS w trybie embedded. Dodatkowo umożliwia to single binary deployment.
• Jest to przydatne nie tylko na platformach bez sieci (wasm), lecz także w aplikacjach offline-first.

Przykład z oficjalnej dokumentacji

 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    // Inicjalizacja nowego serwera z opcjami
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Uruchomienie serwera w goroutine
22    go ns.Start()
23
24    // Oczekiwanie na gotowość serwera do przyjmowania połączeń
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Połączenie z serwerem
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Subskrypcja tematu
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Wypisanie danych wiadomości
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Zamknięcie serwera (opcjonalne)
45        ns.Shutdown()
46    })
47
48    // Publikacja danych do tematu
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Oczekiwanie na zamknięcie serwera
52    ns.WaitForShutdown()
53}

• Jest to przykład Embedded NATS zamieszczony w oficjalnej dokumentacji NATS, jednak jeśli postępuje się zgodnie z tym przykładem, komunikacja nie odbywa się w trybie 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

• Monitorując ruch sieciowy kierowany na localhost (127.0.0.1) za pomocą polecenia watch 'netstat -an | grep 127.0.0.1' i uruchamiając plik go poleceniem go run ., można zauważyć, że pojawiają się nowe żądania sieciowe wychodzące z domyślnego portu NATS, czyli 4222.

Właściwe konfiguracje dla trybu embedding

• Aby komunikacja odbywała się w zamierzonym trybie embedded, potrzebne są dwie następujące opcje.

  • Klient: należy przekazać opcję InProcessServer.
  • Serwer: w Server.Options należy jawnie ustawić flagę DontListen na true.

• Te elementy nie zostały oficjalnie udokumentowane, a początki tej funkcjonalności można odnaleźć w PR.

  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

Działający przykład trybu 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		// Konfiguracja osadzonego serwera NATS
14		// ustaw DontListen na true
15		DontListen: true,
16	}
17
18	// Inicjalizacja nowego serwera z opcjami
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Uruchomienie serwera w goroutine
26	go ns.Start()
27
28	// Oczekiwanie na gotowość serwera do przyjmowania połączeń
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Połączenie z serwerem przez połączenie in-process
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	// Subskrypcja tematu
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Wypisanie danych wiadomości
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Zamknięcie serwera (opcjonalne)
49		ns.Shutdown()
50	})
51
52	// Publikacja danych do tematu
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Oczekiwanie na zamknięcie serwera
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

• Teraz widać, że nie występuje dodatkowy hop sieciowy, zgodnie z zamierzeniem.

Pod maską

TL;DR

• Jest to diagram sekwencji pokazujący, jakie funkcje i w jaki sposób działają wewnętrznie po uruchomieniu tego kodu w main.go. Zasadnicze punkty są następujące.
  • Dzięki DontListen: true serwer pomija fazę nasłuchiwania klientów zwaną AcceptLoop.
  • Jeśli w opcjach połączenia klienta aktywowana jest opcja InProcessServer, tworzone jest połączenie in-memory: za pomocą net.Pipe powstaje potok, a jego drugi koniec zwracany jest klientowi jako typ net.Conn.
  • Klient i serwer prowadzą komunikację in-process przez to połączenie.

Serwer

AcceptLoop

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

• Najpierw, gdy DontListen ma wartość true, serwer pomija fazę nasłuchiwania klientów zwaną 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}

• Dla informacji: funkcja AcceptLoop wykonuje następujące czynności. Części związane z komunikacją sieciową, takie jak TLS czy hostPort, są zbędne w przypadku komunikacji in-process, więc można je pominąć.

Klient

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}

• Funkcja Connect, która nawiązuje połączenie między serwerem a klientem NATS, przyjmuje URL klienta oraz opcje połączenia. Struktura Options, zawierająca te opcje, posiada pole InProcessServer typu interfejsowego InProcessConnProvider.

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))

• Gdy klient NATS wywołuje Connect i przekazuje nats.InProcessServer(ns) jako wartość pola 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}

• opcja InProcessServer zostaje zastąpiona osadzonym serwerem NATS,

 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}

• Ten interfejs jest wykorzystywany w funkcji createConn, która tworzy połączenie. Jeśli opcja InProcessServer nie jest nil (jest poprawna), wywoływana jest metoda InProcessConn z tej opcji,

 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}

• Wywołuje ona zaimplementowaną w serwerze funkcję InProcessConn.
• Ta funkcja jest wywoływana przez nats.go, czyli bibliotekę klienta Go, gdy opcja InProcessServer obiektu nc (połączenia NATS) nie jest nil. Tworzy ona połączenie (net.Conn) i wiąże je z połączeniem serwera.

Interfejsy definiowane przez konsumenta w Go

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.

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.

Generality, Effective Go

• Ten projekt interfejsu dobrze ilustruje pojęcia, o których w Go mówi się często: interfejs definiowany przez konsumenta (consumer defined interface) oraz typowanie strukturalne (structural typing, duck typing). Dlatego chciałbym omówić także ten temat.

 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}

• Wróćmy do kodu. W kliencie nats.go pole opcji InProcessServer zostało zdefiniowane jako interfejs InProcessConnProvider, który obejmuje wyłącznie metodę 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}

• Jednak typem, który do niego podstawiono, jest Server z nats-server, który wykonuje wiele innych funkcji poza InProcessConn.
• Wynika to z tego, że w tej sytuacji klienta interesuje jedynie to, czy dostarczono interfejs InProcessConn, a pozostałe elementy nie są istotne.
• Dlatego klient nats.go używa wyłącznie interfejsu InProcessConnProvider, zdefiniowanego jako consumer defined interface, który zawiera jedynie funkcję InProcessConn() (net.Conn, error).

Podsumowanie

• Krótko omówiono tryb embedded NATS, sposób jego działania oraz interfejsy definiowane przez konsumenta w Go, które można zaobserwować w kodzie NATS.
• Mam nadzieję, że ta informacja okaże się pomocna dla osób korzystających z NATS w podobnym celu, i na tym kończę artykuł.