Files
dingtalk-workspace-cli/internal/event/transport/frame.go
T
aa6abc5ed6 feat(event): add personal event subscriptions (#589)
* dws event

* fix consume fail

* test: add stream ticket injection probe

* feat: add portal ticket stream mode

* user event

* fix: allow portal ticket normal without app secret

* event

* user event

* eventType filter

* refactor(event): 优化IPC端点路径处理和改进相关测试

- 用dwsevent.IPCEndpoint替代原先根据GOOS判断的路径逻辑
- 新增event包实现Unix socket路径长度限制及长路径fallback机制
- 添加endpoint_test.go覆盖路径短长及唯一性的单元测试
- 修改busctl模块使用统一的IPC端点获取方法,避免重复实现
- transport_unix.go新增checkSocketPath函数检查路径长度,防止EINVAL错误
- 在监听和连接Unix socket时加入路径限制检查,提升错误明晰度
- 去除多个文件中无用的runtime导入,简化代码依赖

* opt

* event skill

* default value

* install script event

* fix: remove subscribe id event fanout filter

* fix(personal): 修正指定发送人消息描述错误

* more im event

* filter subId

* fix: align personal event schema with stream payload

* fix: avoid duplicate app helper name

* feat: simplify personal event schemas

* feat: simplify event schema output

* docs: refine dingtalk event skill references

* feat: align personal event consume flags

* fix event stop and status visibility

* hide app event public entrypoints

* hide incomplete personal sender event

* remove external event reference comments

* chore(event): prepare official release

* fix(event): harden personal stream lifecycle

---------

Co-authored-by: 玉澜 <yulan.wqy@alibaba-inc.com>
Co-authored-by: zhengyubai <zhengyubai618@gmail.com>
2026-07-10 17:54:43 +08:00

118 lines
4.1 KiB
Go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package transport
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"io"
)
// MaxFrameBytes caps a single frame's wire size. 1 MiB accommodates large
// event payloads (kanban cards, doc snapshots) with headroom while still
// preventing a buggy/malicious peer from causing OOM via an unbounded line.
// Callers writing frames larger than this should split their work, not bump
// the cap — the SDK enforces its own cloud-side cap that is well below this.
const MaxFrameBytes = 1 << 20 // 1 MiB
// ErrFrameTooLarge is returned by Reader.Read when an incoming frame exceeds
// MaxFrameBytes. The underlying connection is left in an indeterminate
// state (the rest of the oversized frame is NOT drained) — callers should
// close the connection on receipt of this error.
var ErrFrameTooLarge = errors.New("transport: frame exceeds MaxFrameBytes")
// Reader reads \n-delimited JSON frames from an underlying byte stream.
// Wrap each connection (one per direction) in a Reader; Reader is not
// safe for concurrent use.
type Reader struct {
br *bufio.Reader
}
// NewReader returns a Reader with a buffer sized to MaxFrameBytes so a
// single frame can fit in the buffer without growing.
func NewReader(r io.Reader) *Reader {
return &Reader{br: bufio.NewReaderSize(r, MaxFrameBytes)}
}
// Read returns the next frame's raw bytes (the terminating \n is stripped).
// Returns io.EOF cleanly when the peer closes the connection between
// frames. Returns ErrFrameTooLarge when a single frame exceeds the cap.
func (r *Reader) Read() ([]byte, error) {
line, err := r.br.ReadSlice('\n')
if err == nil {
// Strip trailing \n. Safe because ReadSlice always includes the
// delimiter when err == nil.
out := make([]byte, len(line)-1)
copy(out, line[:len(line)-1])
return out, nil
}
if errors.Is(err, bufio.ErrBufferFull) {
// Read past current contents (without retaining) to skip the
// remainder of the oversize frame would be the friendly path,
// but plan says callers should close the connection — keep it
// simple and surface ErrFrameTooLarge immediately.
return nil, ErrFrameTooLarge
}
// EOF on a partial line: report it as EOF (clean close between frames
// returns a fresh EOF with an empty line, already handled above).
if errors.Is(err, io.EOF) && len(line) == 0 {
return nil, io.EOF
}
if err != nil {
return nil, err
}
return nil, nil // unreachable
}
// ReadJSON decodes the next frame into dst. dst must be a non-nil pointer.
func (r *Reader) ReadJSON(dst any) error {
raw, err := r.Read()
if err != nil {
return err
}
if err := json.Unmarshal(raw, dst); err != nil {
return fmt.Errorf("transport: decode frame: %w", err)
}
return nil
}
// Writer marshals values to JSON and writes them as \n-terminated frames.
// Writer is not safe for concurrent use; callers wanting a fan-in writer
// must serialise externally.
type Writer struct {
w io.Writer
}
// NewWriter returns a Writer wrapping w. The Writer does NOT buffer —
// every WriteJSON call performs a single Write to the underlying stream
// so backpressure is immediate.
func NewWriter(w io.Writer) *Writer { return &Writer{w: w} }
// WriteJSON serialises v and writes it as one frame. Refuses to write
// frames larger than MaxFrameBytes with ErrFrameTooLarge.
func (w *Writer) WriteJSON(v any) error {
buf, err := json.Marshal(v)
if err != nil {
return fmt.Errorf("transport: encode frame: %w", err)
}
if len(buf)+1 > MaxFrameBytes {
return ErrFrameTooLarge
}
buf = append(buf, '\n')
_, err = w.w.Write(buf)
return err
}