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
47 lines
1.4 KiB
Go
47 lines
1.4 KiB
Go
// Copyright (c) 2026 Lark Technologies Pte. Ltd.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package consume
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
"time"
|
|
|
|
"github.com/larksuite/cli/internal/event/protocol"
|
|
)
|
|
|
|
const helloAckTimeout = 5 * time.Second // symmetric with bus-side hello read deadline
|
|
|
|
// doHello returns a bufio.Reader holding any bytes already pulled off conn so events
|
|
// buffered with the ack in one TCP segment aren't dropped.
|
|
func doHello(conn net.Conn, eventKey string, eventTypes []string, subscriptionID string) (*protocol.HelloAck, *bufio.Reader, error) {
|
|
hello := protocol.NewHello(os.Getpid(), eventKey, eventTypes, "v1", subscriptionID)
|
|
if err := protocol.EncodeWithDeadline(conn, hello, protocol.WriteTimeout); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
if err := conn.SetReadDeadline(time.Now().Add(helloAckTimeout)); err != nil {
|
|
return nil, nil, fmt.Errorf("set hello_ack deadline: %w", err)
|
|
}
|
|
br := bufio.NewReader(conn)
|
|
line, err := protocol.ReadFrame(br)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("no hello_ack received: %w", err)
|
|
}
|
|
// best-effort clear; if the conn is already broken, the loop's first read will surface it
|
|
_ = conn.SetReadDeadline(time.Time{})
|
|
msg, err := protocol.Decode(bytes.TrimRight(line, "\n"))
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
ack, ok := msg.(*protocol.HelloAck)
|
|
if !ok {
|
|
return nil, nil, fmt.Errorf("expected hello_ack, got %T", msg)
|
|
}
|
|
return ack, br, nil
|
|
}
|