mesh

README

sup/mesh

sup/mesh provides a small set of NATS-backed actors built on top of the sup actor library. The package exposes a single Actor type that manages its own NATS connection and can both publish and subscribe to subjects. The actor's responsibility is connection lifecycle and supervision — subscription handlers are executed directly by the NATS client and should manage their own concurrency if needed.

Installation

go get github.com/webermarci/sup/mesh

Quick Start

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/webermarci/sup"
	"github.com/webermarci/sup/mesh"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	options := nats.GetDefaultOptions()
	options.Url = "nats://localhost:4222"
	
	actor := mesh.NewActor("node", options,
		mesh.WithSubscription("devices.updates", func(b []byte) { 
			fmt.Println("received:", string(b)) 
		}),
		mesh.WithQueueSubscription("devices.updates", "worker-group", func(b []byte) { 
      fmt.Println("received in worker group:", string(b)) 
    }),
	)

	supervisor := sup.NewSupervisor("root",
		sup.WithActors(actor),
		sup.WithPolicy(sup.Permanent),
		sup.WithRestartDelay(time.Second),
	)

	go supervisor.Run(ctx)

	// Give the connection a moment to establish
	time.Sleep(250 * time.Millisecond)

	if err := actor.Publish("topic", []byte("hello mesh")); err != nil {
		fmt.Println("publish error:", err)
	}
}

Documentation

Index

• type Actor
  • func NewActor(name string, options nats.Options, opts ...ActorOption) *Actor
  • func (a *Actor) Publish(subject string, data []byte) error
  • func (a *Actor) Run(ctx context.Context) error
• type ActorOption
  • func WithQueueSubscription(subject, queue string, handler func(*nats.Msg)) ActorOption
  • func WithSubscription(subject string, handler func(*nats.Msg)) ActorOption

Types

type Actor

type Actor struct {
	*sup.BaseActor
	// contains filtered or unexported fields
}

Actor represents a NATS client that can publish messages and subscribe to subjects.

func NewActor

func NewActor(name string, options nats.Options, opts ...ActorOption) *Actor

NewActor creates a new Actor with the given name, NATS options, and optional ActorOptions.

func (*Actor) Publish

func (a *Actor) Publish(subject string, data []byte) error

Publish sends a message with the given data to the specified subject.

func (*Actor) Run

func (a *Actor) Run(ctx context.Context) error

Run starts the Actor's main loop, establishing a NATS connection and setting up subscriptions.

type ActorOption

type ActorOption func(*Actor)

func WithQueueSubscription

func WithQueueSubscription(subject, queue string, handler func(*nats.Msg)) ActorOption

WithQueueSubscription adds a queue-group subscription to the Actor for the given subject, queue, and handler.

func WithSubscription

func WithSubscription(subject string, handler func(*nats.Msg)) ActorOption

WithSubscription adds a subscription to the Actor for the given subject and handler.