integrations

🦾 /pkg/integrations

cd /

Note

asyncmachine-go is a declarative execution model based on AOP, actor model, and state machines.

/pkg/integrations integrates asyncmachine with other projects and technologies:

• JSON
• NATS
• MCP
• Interpreted State Machines

JSON

JSON types cover mutations, subscriptions, and data getters. Each of these is divided in request and response objects which have a (/docs/jsonschema). Their usage depends on the specific implementation, eg in NATS each machine has a dedicated subtopic for mutation requests.

• MutationReq (jsonschema)
• MutationResp (jsonschema)
• WaitingReq (jsonschema)
• WaitingResp (jsonschema)
• GetterReq (jsonschema)
• GetterResp (jsonschema)

import amjson "github.com/pancsta/asyncmachine-go/pkg/integration"

// create a subscription to Foo
reqSub := integrations.NewWaitingReq()
reqSub.States = am.S{"Foo"}
j, err := json.Marshal(reqSub)

NATS

NATS is a popular and high-performance messaging system made in Go. State machines are exposed under a topic, with each state machine also being subscribed to a dedicated subtopic "[topic].[machineID]" for mutation requests. Optional [queue] allows to load-balance requests across multiple subscribers.

import am "github.com/pancsta/asyncmachine-go"
import nats "github.com/pancsta/asyncmachine-go/pkg/integration/nats"

// ...

// var mach *am.Machine
// var ctx context.Context
// var nc *nats.Conn

// expose mach under mytopic
_ = nats.ExposeMachine(ctx, mach, nc, "mytopic", "")
// mutate - add Foo
res, _ := nats.Add(ctx, nc, topic, mach.Id(), am.S{"Foo"}, nil)
if res == am.Executed {
    print("Foo added to mach")
}

TODO

• recipient matching (filters similar to the REPL ones)
• better error handling (avoid overreporting)

MCP

Every state machine with typed args (including network machines) can have an MCP server thanks to mcp-go. See /tools/cmd/am-dbg for how to extend it with custom getters (as asyncmachine does not return data).

import ammcp "github.com/pancsta/asyncmachine-go/pkg/integrations/mcp"

// ...

var mach am.Api

srv, err := ammcp.New(mach, ammcp.Opts{
    Name: "am-dbg",
    Desc: "MCP to control the asyncmachine-go TUI debugger." +
        " The double-line border panel is currently focused. Most features are accessible via ToggleTool.",
    MutCallback: func(ctx context.Context) error {
        ok := d.Mach.Eval("MCPTool", func() {
            d.hRedrawFull(false)
        }, ctx)
        if !ok {
            return am.ErrEvalTimeout
        }
        return nil
    },
    StatesInclude:  states.DebuggerGroups.Mcp,
    StatesReadonly: states.DebuggerGroups.McpReadonly,
    Args:           types.ARpc{},
    ParseRpc:       types.ParseRpc,
    StateCalls:     types.StateCalls,
    Version:        d.params.Version,
})
if err != nil {
    return nil, err
}
srv.Http.Start(":8753")

Interpreted State Machines

State machines can be extended from interpreted (dynamic) Go source code without compilation via traefik/yaegi. This includes adding new states, modifying schema relations, and binding handlers. It can happen either inside the interpreted code or outside in the host. The host is available as H and the Run functions return extensions of the passed asyncmachine. See /pkg/integrations/yaegi.

type Host struct {
	Mach *am.Machine
	Host any
}

var H *Host

type Ret struct {
	Schema    am.Schema
	Names     am.S
	BindingId string
}

Status

Alpha, work in progress, not semantically versioned.

monorepo

Go back to the monorepo root to continue reading.