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
43 lines
1.0 KiB
Go
43 lines
1.0 KiB
Go
// Copyright (c) 2026 Lark Technologies Pte. Ltd.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package consume
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"net"
|
|
"time"
|
|
|
|
"github.com/larksuite/cli/internal/event/protocol"
|
|
)
|
|
|
|
const preShutdownAckTimeout = 2 * time.Second
|
|
|
|
// checkLastForKey atomically reserves a cleanup lock; on any error defaults to true
|
|
// (cleanup-on-error is safer than leaking server state). Discards non-ack frames in flight.
|
|
func checkLastForKey(conn net.Conn, eventKey string, subscriptionID string) bool {
|
|
msg := protocol.NewPreShutdownCheck(eventKey, subscriptionID)
|
|
if err := protocol.EncodeWithDeadline(conn, msg, protocol.WriteTimeout); err != nil {
|
|
return true
|
|
}
|
|
|
|
if err := conn.SetReadDeadline(time.Now().Add(preShutdownAckTimeout)); err != nil {
|
|
return true
|
|
}
|
|
br := bufio.NewReader(conn)
|
|
for {
|
|
line, err := protocol.ReadFrame(br)
|
|
if err != nil {
|
|
return true
|
|
}
|
|
resp, err := protocol.Decode(bytes.TrimRight(line, "\n"))
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if ack, ok := resp.(*protocol.PreShutdownAck); ok {
|
|
return ack.LastForKey
|
|
}
|
|
}
|
|
}
|