イベントバス

イベントバスは、キューに入った pub/sub アクションを 1 つの dispatcher goroutine で処理し、一致するイベントを subscriber channel へ配信します。

Go のスニペットは実装および拡張の断片です。既存のコンポーネント context、logger、handler、アプリケーションイベント型を前提としています。

イベント構造

type Event struct {
    System string  // Component/module (e.g., "registry", "process")
    Kind   string  // Event type (e.g., "create", "update", "exit")
    Path   string  // Entity identifier
    Data   any     // Payload
    Aux    any     // In-process dispatcher context; not propagated to processes
}

バスアーキテクチャ

flowchart LR
    subgraph Publishers
        P1[Component]
        P2[Component]
    end

    subgraph Bus
        Q[actionQueue]
        D[dispatcher goroutine]
        S[subscribers map]
    end

    subgraph Subscribers
        S1[chan Event]
        S2[chan Event]
    end

    P1 & P2 -->|enqueue| Q
    Q -->|signal| D
    D -->|match & deliver| S1 & S2
    D <-->|manage| S

バスは単純な構造体に状態を格納します。

type Bus struct {
    subscribers       map[SubscriberID]sub
    subscriberCounter uint64
    maxSubscribers    int

    actionQueue []action
    spareQueue  []action
    actionMu    sync.Mutex
    actionReady chan struct{}  // buffered=1

    closed atomic.Bool
}

すべての変更は dispatcher goroutine を経由するため、複雑なロックなしで競合状態を排除できます。

アクション

キューには 4 種類のアクションが流れます。

アクション	動作
Subscribe	subscriber をマップへ追加し、done channel へ応答
Unsubscribe	subscriber を削除し、done channel へ応答
Send	一致する subscriber へイベントを配信
Stop	subscriber を消去し、キューを drain してループを終了

Subscribe と Unsubscribe は dispatcher の確認までブロックします。Send は fire-and-forget です。バスは最大 DefaultMaxSubscribers 件（デフォルト 4096）の subscription を受け付けます。上限を超えた subscription は ErrSubscribersCapReached で失敗します。

バスがDefaultMaxSubscribers（4096）個のアクティブなサブスクリプションを保持すると、SubscribeはErrSubscribersCapReachedで拒否される。

Subscribeは、サブスクリプションのコンテキストが既にキャンセルされている場合は即座に失敗し、所有権の判断が下される前にキャンセルされた場合はディスパッチャ側で再度失敗する。バスは自身がインストールしていないチャネルを決して受け取らない。

Unsubscribeはベストエフォートのヒントではなく、所有権のバリアである。ディスパッチャが確認応答した後にのみ戻るため、呼び出し側はバスが送信中の参照を保持していないことを前提にチャネルを解放できる。Stopの後に到着した場合、確認応答はディスパッチャが既にドレイン済みのバッチの配信を終えるまで待機する。

Stopも同様に終端的である。並行する2回目のStopは、クローズ済みフラグを見て早期に戻ることはなく、ディスパッチャがドレインして終了するまで待機する。

キュースワッピング

dispatcher は、定常状態での allocation を避けるためスライスを交換します。

func (b *Bus) processActions() bool {
    b.actionMu.Lock()
    actions := b.actionQueue
    b.actionQueue = b.spareQueue[:0]
    b.spareQueue = nil
    b.actionMu.Unlock()

    for i := range actions {
        // process action
    }

    clear(actions)
    b.actionMu.Lock()
    b.spareQueue = actions[:0]
    b.actionMu.Unlock()
    return true
}

2 つのスライスが交互に使われます。一方は処理用、もう一方は新規到着用です。actionReady channel の buffer は 1 なので、signal はブロックせず、複数の enqueue は 1 回の wakeup にまとめられます。

パターンマッチング

subscription は subscribe 時に一度だけパターンをコンパイルします。

type sub struct {
    subID   SubscriberID
    ctx     context.Context
    system  *wildcard.Wildcard
    kind    *wildcard.Wildcard
    eventCh chan<- Event
}

wildcard パッケージは 4 種類のパターンに対応します。

パターン	一致対象
`registry`	完全一致のみ
`*`	任意の 1 セグメント
`**`	0 個以上のセグメント
`(a\|b)`	セグメント内の選択肢

パターンは . で分割されるため、registry.* は registry.create に一致しますが、registry.entry.create には一致しません。パターン registry.** は registry、registry.create、registry.entry.create のすべてに一致します。

イベント配信

Send の処理中、dispatcher は subscriber を反復処理します。

for id, s := range b.subscribers {
    if s.system != nil && !s.system.Match(a.event.System) {
        continue
    }
    if s.kind != nil && !s.kind.Match(a.event.Kind) {
        continue
    }

    select {
    case <-a.ctx.Done():
        goto cleanup
    case <-s.ctx.Done():
        expiredSubs = append(expiredSubs, id)
    case s.eventCh <- a.event:
    }
}

subscriber の context が cancel されている場合、その配信 pass 中に削除対象としてマークされます。イベントの context によって反復処理の途中で配信を cancel することもできます。

Lua プロセスブリッジ

events dispatcher は Go イベントを Lua プロセスへ bridge します。すべてのイベント（"**"）を一度 subscribe し、プロセスの subscription に基づいて内部ルーティングします。

type Dispatcher struct {
    bus    event.Bus
    node   relay.Node
    subID  SubscriberID
    eventC chan event.Event

    mu   sync.RWMutex
    subs map[string]*subscription  // topic -> subscription
}

Lua プロセスが events.subscribe() を介して subscribe すると、dispatcher はパターンと対象 PID を格納します。一致するイベントは package 化され、relay 経由で送信されます。

func (d *Dispatcher) routeEvent(evt event.Event) {
    d.mu.RLock()
    defer d.mu.RUnlock()

    for _, sub := range d.subs {
        if !matchPattern(sub.system, evt.System) {
            continue
        }
        if sub.kind != "" && sub.kind != "*" && !matchPattern(sub.kind, evt.Kind) {
            continue
        }

        data := map[string]any{
            "system": evt.System,
            "kind":   evt.Kind,
            "path":   evt.Path,
        }
        if evt.Data != nil {
            data["data"] = evt.Data
        }

        pkg := relay.NewPackage(pid.PID{}, sub.pid, sub.topic, payload.New(data))
        d.node.Send(pkg)
    }
}

ヘルパー型

Subscriber

channel subscription を callback でラップします。

handler, err := eventbus.NewSubscriber(ctx, bus, "registry", "entry.*",
    func(evt Event) {
        // handle
    })
if err != nil {
    return err
}
defer handler.Close()

2 つの goroutine を生成します。1 つはイベントを読み取って handler を呼び出し、もう 1 つは context の cancel を待って unsubscribe します。

EventRouter

一元化されたライフサイクルで複数の handler を管理します。

router, err := eventbus.StartRouter(ctx, bus,
    WithHandlers(handler1, handler2),
    WithLogger(log))
if err != nil {
    return err
}
defer router.Stop()

各 handler は Pattern() と Handle() を実装します。router は handler ごとに Subscriber を作成し、Stop 時にすべて閉じます。

AwaitService

pub/sub上でのリクエスト・レスポンス。(system, kind)ペアごとに単一のサブスクリプションを保持し、Pathによってイベントをwaiterにルーティング：

svc := eventbus.NewAwaitService(bus)
svc.Start(ctx)
defer svc.Stop()

waiter, _ := svc.Prepare(ctx, "test", "response.(accept|reject)", "test/path", 5*time.Second)
defer waiter.Close()

bus.Send(ctx, triggeringEvent)

result := waiter.Wait()  // AwaitResult{Event, Accepted, Error}を返す

Prepareはトリガーとなるイベントを送信する前にwaiterを登録し、待機の登録前にレスポンスが到着する競合状態を回避する。WaitはPathがマッチするイベントの到着、またはタイムアウト（非正の値の場合はデフォルトのDefaultAwaitTimeout、30秒）の満了までブロック。Acceptedはイベント種別がaccept、*.accept、*.acceptedのいずれかの場合にtrueとなり、それ以外の種別は拒否として扱われ、Data内のerrorはErrorとして返される。便宜的なAwait(ctx, system, kind, path, timeout)はPrepareとWaitを組み合わせたもの。ブートインフラストラクチャはAwaitServiceをコンテキストに登録する（event.GetAwaitService）。

シャットダウン

Stop() が closed フラグを atomic に設定し、Stop アクションを enqueue
Dispatcher が subscriber map を消去
残りのキュー済みアクションを drain
- Subscribe リクエストは「bus is closed」エラーを受信
- Unsubscribe リクエストは即座に完了
- Send イベントは破棄
WaitGroup が完了