---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Konfigurácia embedded NATS servera v aplikáciách Go a štruktúra komunikácie in-process
description: Tento text sa zaoberá konfiguráciou a spustením NATS brokera (nats-server/v2) vnoreného priamo v rámci procesu Go, ako aj návrhom inmemory komunikácie založenej na InProcessConn a rozhraním definovaným spotrebiteľom.
language: sk
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Začíname

## O NATS

> Softvérové aplikácie a služby potrebujú vymieňať dáta. NATS je infraštruktúra, ktorá umožňuje takúto výmenu dát, segmentovanú vo forme správ. Nazývame to „**message oriented middleware**“.
>
> S NATS môžu vývojári aplikácií:
>
> - Bez námahy budovať distribuované a škálovateľné klient-serverové aplikácie.
> - Ukladať a distribuovať dáta v reálnom čase všeobecným spôsobom. Toto je možné flexibilne dosiahnuť naprieč rôznymi prostrediami, jazykmi, poskytovateľmi cloudu a lokálnymi systémami.
>
> [Čo je NATS, dokumentácia NATS](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS je message broker napísaný v jazyku Go.

## Embedded NATS

> Ak je vaša aplikácia napísaná v Go a ak to zodpovedá vášmu prípadu použitia a scenárom nasadenia, môžete dokonca vložiť server NATS priamo do svojej aplikácie.

[Embedding NATS, dokumentácia NATS](https://docs.nats.io/running-a-nats-service/clients#embedding-nats)

>

- A NATS má aj jednu osobitosť: pre aplikácie napísané v Go podporuje embedded mode.
- Inými slovami, namiesto bežného spôsobu, pri ktorom sa samostatný server brokera spustí a aplikácia s ním komunikuje prostredníctvom klienta, je možné samotný broker vložiť (embed) priamo do aplikácie napísanej v Go.

## Výhody a prípady použitia vloženého NATS

- Keďže existuje dobre vysvetľujúce [video na Youtube](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN), odkazujem na toto video namiesto podrobného výkladu.
- Aj bez nasadenia samostatného servera message brokera je možné vytvoriť modulárnu monolitickú aplikáciu (modular monolith) a dosiahnuť separation of concerns, pričom sa zároveň využije možnosť vložiť NATS ako embedded. Navyše je umožnené nasadenie ako jediný binárny súbor (single binary deployment).
- Okrem platforiem bez sieťového pripojenia (wasm) je to užitočné aj v aplikáciách typu offline-first.

# Príklad z oficiálnej dokumentácie

```go
package main

import (
    "fmt"
    "time"

    "github.com/nats-io/nats-server/v2/server"
    "github.com/nats-io/nats.go"
)

func main() {
    opts := &server.Options{}

    // Inicializácia nového servera s možnosťami
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // Spustenie servera cez goroutine
    go ns.Start()

    // Počkajte, kým bude server pripravený na pripojenia
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Pripojenie k serveru
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Prihlásenie na odber témy
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Vypísanie dát správy
        data := string(msg.Data)
        fmt.Println(data)

        // Vypnutie servera (voliteľné)
        ns.Shutdown()
    })

    // Publikovanie dát do témy
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Čakanie na vypnutie servera
    ns.WaitForShutdown()
}
```

- Toto je príklad vloženého NATS, na ktorý odkazuje oficiálna dokumentácia NATS, no ak sa budeme riadiť týmto príkladovým kódom, komunikácia neprebieha v embedding mode.

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

- Pomocou príkazu `watch 'netstat -an | grep 127.0.0.1'` sledujeme sieťovú komunikáciu smerujúcu na [localhost](http://localhost) (127.0.0.1) a pri spustení tohto go súboru príkazom `go run .` vidíme, ako pribúdajú nové sieťové požiadavky vychádzajúce z predvoleného portu NATS `4222`.

# Správne konfigurácie pre embedding mode

- Aby komunikácia prebiehala v embedded mode tak, ako je zamýšľané, sú potrebné dve nasledujúce možnosti:
  - Klient: je potrebné nastaviť option `InProcessServer`.
  - Server: vo `Server.Options` je potrebné explicitne nastaviť flag `DontListen` na `true`.
- Tieto časti nie sú oficiálne zdokumentované a začiatok tejto funkcionality je možné vystopovať v tomto [PR](https://github.com/nats-io/nats-server/pull/2360).

  > Tento PR pridáva tri veci:
  >
  > 1. funkcia `InProcessConn()` pre `Server`, ktorá vytvára `net.Pipe` na získanie pripojenia k serveru NATS bez použitia TCP socketov
  > 2. option `DontListen`, ktorá serveru NATS hovorí, aby nepočúval na obvyklom TCP listeneri
  > 3. kanál `startupComplete`, ktorý sa zatvorí tesne pred spustením `AcceptLoop`, a `readyForConnections` naň bude čakať
  >
  > Hlavnou motiváciou je, že máme aplikáciu, ktorá môže bežať buď v monolitickom (jednoprocesovom) režime, alebo v polylitickom (viacprocesovom) režime. Chceli by sme pre zjednodušenie používať NATS v oboch režimoch, no monolitický režim musí podporovať rôzne platformy, kde otváranie socketových pripojení buď nedáva zmysel (mobilné zariadenia), alebo nie je vôbec možné (WASM). Tieto zmeny nám umožnia používať NATS úplne v rámci jedného procesu.
  >
  > Sprievodný PR [nats-io/nats.go#774](https://github.com/nats-io/nats.go/pull/774) pridáva podporu na strane klienta.
  >
  > Toto je môj prvý PR do tohto projektu, takže sa vopred ospravedlňujem, ak som niečo zjavné prehliadol.
  >
  > /cc @nats-io/core

# Pracovný príklad pre embedded mode

```go
package main

import (
	"fmt"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
)

func main() {
	opts := &server.Options{
		// Pre konfiguráciu vloženého servera NATS
		// nastavte DontListen na true
		DontListen: true,
	}

	// Inicializácia nového servera s možnosťami
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// Spustenie servera cez goroutine
	go ns.Start()

	// Počkajte, kým bude server pripravený na pripojenia
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Pripojenie k serveru cez in-process pripojenie
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Prihlásenie na odber témy
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Vypísanie dát správy
		data := string(msg.Data)
		fmt.Println(data)

		// Vypnutie servera (voliteľné)
		ns.Shutdown()
	})

	// Publikovanie dát do témy
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Čakanie na vypnutie servera
	ns.WaitForShutdown()
}

```

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

```

- Teraz vidíme, že nedochádza k žiadnemu dodatočnému sieťovému skoku (network hop), ako sme zamýšľali.

# Pod kapotou

## TL;DR

![diagram1](/assets/images/embedded-nats/seq1.svg)

- Ide o sekvenčný diagram znázorňujúci, aké funkcie sa interne vykonávajú a ako fungujú, keď sa tento kód spustí v `main.go`. Hlavné body sú nasledujúce:
  - Vďaka `DontListen: true` server vynecháva fázu počúvania klientov nazývanú `AcceptLoop`.
  - Ak je v možnostiach pripojenia klienta aktivovaná option `InProcessServer`, vytvorí sa in-memory pripojenie: pomocou `net.Pipe` sa vytvorí rúra a jej koniec sa klientovi vráti ako typ `net.Conn`.
  - Klient a server komunikujú v rámci tohto pripojenia v in-process režime.

## Server

### AcceptLoop

```go
// nats-server/server/server.go

// Čakanie na klientov.
if !opts.DontListen {
	s.AcceptLoop(clientListenReady)
}
```

- Najprv, ak je `DontListen` true, server vynecháva fázu počúvania klientov nazývanú `AcceptLoop`.

```go
// nats-server/server/server.go

// AcceptLoop je exportovaná kvôli jednoduchšiemu testovaniu.
func (s *Server) AcceptLoop(clr chan struct{}) {
	// Ak by sme skončili skôr, než sa listener správne nastaví, zabezpečte zatvorenie kanála.
	defer func() {
		if clr != nil {
			close(clr)
		}
	}()

	if s.isShuttingDown() {
		return
	}

	// Snímka možností servera.
	opts := s.getOpts()

	// Nastavenie stavu, ktorý umožňuje vypnutie
	s.mu.Lock()
	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
	l, e := natsListen("tcp", hp)
	s.listenerErr = e
	if e != nil {
		s.mu.Unlock()
		s.Fatalf("Error listening on port: %s, %q", hp, e)
		return
	}
	s.Noticef("Listening for client connections on %s",
		net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))

	// Upozornenie, že TLS je povolené.
	if opts.TLSConfig != nil {
		s.Noticef("TLS required for client connections")
		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
		}
	}

	// Ak bol server spustený s RANDOM_PORT (-1), opts.Port by sa na začiatku tejto funkcie rovnal 0. Preto musíme získať skutočný port.
	if opts.Port == 0 {
		// Zapíšte získaný port späť do možností.
		opts.Port = l.Addr().(*net.TCPAddr).Port
	}

	// Keď je port nastavený (ak bol nastavený na RANDOM), nastavte Host/Port v info servera buď z hodnôt Options, alebo z ClientAdvertise.
	if err := s.setInfoHostPort(); err != nil {
		s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
		l.Close()
		s.mu.Unlock()
		return
	}
	// Uchovávajte URL adresy na pripojenie klientov. Môžeme ich neskôr potrebovať.
	s.clientConnectURLs = s.getClientConnectURLs()
	s.listener = l

	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
		func(_ error) bool {
			if s.isLameDuckMode() {
				// Signal that we are not accepting new clients
				s.ldmCh <- true
				// Now wait for the Shutdown...
				<-s.quitCh
				return true
			}
			return false
		})
	s.mu.Unlock()

	// Oznámte volajúcemu, že sme pripravení
	close(clr)
	clr = nil
}
```

- Pre úplnosť, funkcia AcceptLoop vykonáva nasledujúce kroky. Ide o časti týkajúce sa sieťovej komunikácie, ako sú `TLS` alebo `hostPort`; pri in-process komunikácii nie sú potrebné, takže ich možno vynechať.

## Client

### InProcessServer

```go

// nats-go/nats.go

// Connect sa pokúsi pripojiť k systému NATS.
// URL môže obsahovať používateľské meno a heslo, napr. nats://derek:pass@localhost:4222
// Podporované sú aj zoznamy oddelené čiarkami, napr. urlA, urlB.
// Možnosti sa najprv nastavia na predvolené hodnoty, ale je možné ich prepísať.
// Na pripojenie k websocket portu servera NATS použite schému `ws` alebo `wss`, napríklad
// `ws://localhost:8080`. Schémy websocket nemožno kombinovať s inými (nats/tls).
func Connect(url string, options ...Option) (*Conn, error) {
	opts := GetDefaultOptions()
	opts.Servers = processUrlString(url)
	for _, opt := range options {
		if opt != nil {
			if err := opt(&opts); err != nil {
				return nil, err
			}
		}
	}
	return opts.Connect()
}
```

```go
// nats-go/nats.go

// Options slúži na vytvorenie prispôsobeného pripojenia.
type Options struct {
	// Url predstavuje jednu URL adresu servera NATS, ku ktorej sa klient pripojí. Ak je nastavená aj option Servers, stane sa prvým serverom v poli Servers.
	Url string

	// InProcessServer predstavuje server NATS bežiaci v rámci rovnakého procesu. Ak je nastavený, pokúsime sa pripojiť priamo k serveru namiesto použitia externých TCP pripojení.
	InProcessServer InProcesConnProvider

	//...
}
```

```go
// nats-go/nats.go

type InProcesConnProvider interface {
	InProcessConn() (net.Conn, error)
}
```

- Funkcia `Connect`, ktorá zabezpečuje pripojenie servera NATS a klienta NATS, prijíma URL klienta a connect options; štruktúra Options, ktorá tieto options zhromažďuje, obsahuje pole `InProcessServer` typu rozhrania `InProcesConnProvider`.

```go
// main.go of example code

// Inicializácia nového servera s možnosťami
ns, err := server.NewServer(opts)

//...

// Pripojenie k serveru cez in-process pripojenie
nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
```

- Keď klient NATS volá Connect a ako pole `InProcessServer` odovzdá `nats.InProcessServer(ns)`,

```go
// nats-go/nats.go

// InProcessServer je Option, ktorá sa pokúsi nadviazať priame spojenie so serverom NATS bežiacim v rámci procesu namiesto vytáčania cez TCP.
func InProcessServer(server InProcesConnProvider) Option {
	return func(o *Options) error {
		o.InProcessServer = server
		return nil
	}
}
```

- hodnota option InProcessServer sa nahradí embedded serverom NATS,

```go
// nats-go/nats.go

// createConn sa pripojí k serveru a obalí zodpovedajúce štruktúry bufio. Pri existujúcom pripojení sa správne zachová.
func (nc *Conn) createConn() (err error) {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	}

	// Ak máme referenciu na in-process server, nadviažeme pripojenie cez ňu.
	if nc.Opts.InProcessServer != nil {
		conn, err := nc.Opts.InProcessServer.InProcessConn()
		if err != nil {
			return fmt.Errorf("failed to get in-process connection: %w", err)
		}
		nc.conn = conn
		nc.bindToNewConn()
		return nil
	}

	//...
}
```

- toto rozhranie sa v funkcii `createConn`, ktorá vytvára pripojenie, použije tak, že ak option `InProcessServer` nie je nil (je platná), spustí `InProcesConn` z InProcessServer uvedeného v option
- a zavolá implementáciu `InProcessConn` na strane servera.

```go
// nats-server/server/server.go

// InProcessConn vráti in-process pripojenie k serveru a eliminuje potrebu použitia TCP listeneru na lokálnu konektivitu v rámci toho istého procesu. Môže sa použiť bez ohľadu na stav option DontListen.
func (s *Server) InProcessConn() (net.Conn, error) {
	pl, pr := net.Pipe()
	if !s.startGoRoutine(func() {
		s.createClientInProcess(pl)
		s.grWG.Done()
	}) {
		pl.Close()
		pr.Close()
		return nil, fmt.Errorf("failed to create connection")
	}
	return pr, nil
}
```

- Táto funkcia sa volá v go klientovi NATS, `nats.go`, keď `InProcessServer` nc (pripojenia NATS) nie je nil; vytvorí pripojenie (`net.Conn`) a zviaže ho s pripojením servera.

# Consumer driven interface v jazyku Go

> Typ implementuje rozhranie tým, že implementuje jeho metódy. Neexistuje žiadne explicitné deklarovanie úmyslu, žiadne kľúčové slovo „implements“. Implicitné rozhrania oddeľujú definíciu rozhrania od jeho implementácie, ktorá sa potom môže objaviť v ľubovoľnom balíku bez predchádzajúcej dohody.

[Rozhrania sú implementované implicitne, A Tour of Go](https://go.dev/tour/methods/10)

> Ak typ existuje len preto, aby implementoval rozhranie, a nikdy nebude mať exportované metódy nad rámec tohto rozhrania, nie je potrebné exportovať samotný typ.

[Všeobecnosť, Effective Go](https://go.dev/doc/effective_go#generality)

- Tento návrh rozhraní dobre zachytáva to, čo sa v Go bežne nazýva consumer defined interface a structural typing (duck typing), preto by sme sa k tejto téme chceli vrátiť.

```go
// nats-go/nats.go

// Options slúži na vytvorenie prispôsobeného pripojenia.
type Options struct {
	// Url predstavuje jednu URL adresu servera NATS, ku ktorej sa klient pripojí. Ak je nastavená aj option Servers, stane sa prvým serverom v poli Servers.
	Url string

	// InProcessServer predstavuje server NATS bežiaci v rámci rovnakého procesu. Ak je nastavený, pokúsime sa pripojiť priamo k serveru namiesto použitia externých TCP pripojení.
	InProcessServer InProcesConnProvider

	//...
}
```

```go
// nats-go/nats.go

type InProcesConnProvider interface {
	InProcessConn() (net.Conn, error)
}
```

- Vráťme sa ku kódu. Klient nats.go definuje pole option `InProcessServer` ako rozhranie `InProcesConnProvider`, ktoré vykonáva iba `InProcessConn`.

```go
// nats-server/server/server.go

// InProcessConn vráti in-process pripojenie k serveru a eliminuje potrebu použitia TCP listeneru na lokálnu konektivitu v rámci toho istého procesu. Môže sa použiť bez ohľadu na stav option DontListen.
func (s *Server) InProcessConn() (net.Conn, error) {
	pl, pr := net.Pipe()
	if !s.startGoRoutine(func() {
		s.createClientInProcess(pl)
		s.grWG.Done()
	}) {
		pl.Close()
		pr.Close()
		return nil, fmt.Errorf("failed to create connection")
	}
	return pr, nil
}
```

- Typom, ktorý sa do neho dosadí, je však `Server` z nats-server, ktorý okrem InProcessConn vykonáva aj mnoho ďalších funkcií.
- Dôvod je ten, že v tejto situácii klienta zaujíma len to, či poskytuje rozhranie `InProcessConn`, a na ostatných veciach veľmi nezáleží.
- Klient nats.go preto používa iba consumer defined rozhranie `InProcessConnProvider`, ktoré definuje len funkcionalitu `InProcessConn() (net.Conn, error)`.

# Záver

- Stručne sme sa venovali embedded mode v NATS, jeho fungovaniu a consumer defined rozhraniam v Go, ktoré je možné vidieť v kóde NATS.
- Dúfam, že tieto informácie pomôžu ľuďom, ktorí používajú NATS na podobné účely, a týmto článok končím.