mirror of
https://github.com/larksuite/cli.git
synced 2026-09-14 18:42:53 +08:00
liuxinyanglxy
3f77eded9d
Framework support for resource-scoped event subscriptions, so one EventKey can fan out into independent per-resource subscription scopes: - KeyDefinition gains SubscriptionKey / NormalizeParams / Match hooks - ComputeSubscriptionID derives a dedup identity from (EventKey, sub-key params); plumbed through bus Hub, consume loop, and the Hello / PreShutdownCheck / ConsumerInfo protocol messages - add a synchronous Match filter stage before Process - change PreConsume cleanup to func() error and surface cleanup (unsubscribe) failures as WARN with an idempotency note - adapt minutes/vc/whiteboard PreConsume to the new cleanup signature - render SubscriptionID / SubscriptionKey in event status & schema output No domain wires these hooks yet; covered by unit tests using bus/protocol doubles. (Mail, the original exerciser, is intentionally not included.) Change-Id: Ifc743f1aa0bc4dff0c8a1e35da24883694fe7699
91 lines
1.9 KiB
Go
91 lines
1.9 KiB
Go
// Copyright (c) 2026 Lark Technologies Pte. Ltd.
|
||
// SPDX-License-Identifier: MIT
|
||
|
||
package bus
|
||
|
||
import (
|
||
"io"
|
||
"log"
|
||
"net"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
// Reproduces Run × onClose re-entrant deadlock if b.mu is held across Close.
|
||
func TestRunShutdownWithMultipleConns(t *testing.T) {
|
||
logger := log.New(io.Discard, "", 0)
|
||
hub := NewHub()
|
||
b := &Bus{
|
||
hub: hub,
|
||
logger: logger,
|
||
conns: make(map[*Conn]struct{}),
|
||
}
|
||
|
||
const N = 3
|
||
pipes := make([]net.Conn, 0, N*2)
|
||
t.Cleanup(func() {
|
||
for _, p := range pipes {
|
||
p.Close()
|
||
}
|
||
})
|
||
|
||
for i := 0; i < N; i++ {
|
||
server, client := net.Pipe()
|
||
pipes = append(pipes, server, client)
|
||
|
||
bc := NewConn(server, nil, "im.msg", []string{"im.message.receive_v1"}, 1000+i, "")
|
||
bc.SetLogger(logger)
|
||
hub.RegisterAndIsFirst(bc)
|
||
|
||
bc.SetOnClose(func(c *Conn) {
|
||
b.hub.UnregisterAndIsLast(c)
|
||
b.mu.Lock()
|
||
delete(b.conns, c)
|
||
b.mu.Unlock()
|
||
})
|
||
|
||
b.mu.Lock()
|
||
b.conns[bc] = struct{}{}
|
||
b.mu.Unlock()
|
||
}
|
||
|
||
done := make(chan struct{})
|
||
go func() {
|
||
shutdownConns(b)
|
||
close(done)
|
||
}()
|
||
|
||
select {
|
||
case <-done:
|
||
case <-time.After(2 * time.Second):
|
||
t.Fatal("shutdownConns deadlocked: did not complete within 2s")
|
||
}
|
||
|
||
if got := hub.ConnCount(); got != 0 {
|
||
t.Errorf("expected 0 subscribers in hub after shutdown, got %d", got)
|
||
}
|
||
b.mu.Lock()
|
||
remaining := len(b.conns)
|
||
b.mu.Unlock()
|
||
if remaining != 0 {
|
||
t.Errorf("expected 0 conns in Bus after shutdown, got %d", remaining)
|
||
}
|
||
}
|
||
|
||
// shutdownCh must be buffered so a signal sent before Run's select loop is still delivered.
|
||
func TestShutdownSignalNotDroppedBeforeRunSelects(t *testing.T) {
|
||
b := NewBus("test-app", "test-secret", "", nil, log.New(io.Discard, "", 0))
|
||
|
||
select {
|
||
case b.shutdownCh <- struct{}{}:
|
||
default:
|
||
t.Fatal("handleShutdown's send took default branch — signal would be lost")
|
||
}
|
||
|
||
select {
|
||
case <-b.shutdownCh:
|
||
case <-time.After(200 * time.Millisecond):
|
||
t.Fatal("shutdown signal was not latched")
|
||
}
|
||
}
|