Files
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

132 lines
3.6 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 source
import (
"testing"
"time"
)
func newMachineAt(now time.Time) *Machine {
m := NewMachine()
m.now = func() time.Time { return now }
return m
}
func TestMachine_InitialState(t *testing.T) {
m := NewMachine()
if m.State() != StateDisconnected {
t.Fatalf("initial state = %s, want %s", m.State(), StateDisconnected)
}
snap := m.Snapshot()
if snap.StateSource != SourceInferred {
t.Fatalf("v1 must emit StateSource=inferred, got %s", snap.StateSource)
}
}
func TestMachine_TransitionsHappyPath(t *testing.T) {
now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC)
m := newMachineAt(now)
m.OnConnecting()
if m.State() != StateConnecting {
t.Fatalf("after OnConnecting: %s", m.State())
}
m.OnConnected()
if m.State() != StateConnected {
t.Fatalf("after OnConnected: %s", m.State())
}
m.OnEvent()
if m.State() != StateConnected {
t.Fatalf("after OnEvent: %s (should remain connected)", m.State())
}
m.OnStopped()
if m.State() != StateStopped {
t.Fatalf("after OnStopped: %s", m.State())
}
}
func TestMachine_IdleAfterQuietWindow(t *testing.T) {
now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC)
m := newMachineAt(now)
m.OnConnected()
m.OnEvent() // lastEventAt = now
// Advance time past IdleAfter
m.now = func() time.Time { return now.Add(IdleAfter + 5*time.Second) }
if got := m.State(); got != StateIdle {
t.Fatalf("state after quiet window: %s, want %s", got, StateIdle)
}
// New event resets to connected
m.OnEvent()
if got := m.State(); got != StateConnected {
t.Fatalf("state after new event: %s, want %s", got, StateConnected)
}
}
func TestMachine_DegradedWhenReconnectsButNoEvents(t *testing.T) {
now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC)
m := newMachineAt(now)
m.OnConnected()
m.OnEvent() // baseline
// Advance + reconnect signal (no event recovery)
later := now.Add(IdleAfter + 30*time.Second)
m.now = func() time.Time { return later }
m.OnReconnect() // sets state=reconnecting, lastReconnectAt=later
// State while reconnecting
if got := m.State(); got != StateReconnecting {
t.Fatalf("after OnReconnect: %s, want %s", got, StateReconnecting)
}
// Push state back to connected (as if reconnect succeeded but no traffic)
m.OnConnected()
// Now idle window passes with no events but a recent reconnect → degraded
muchLater := later.Add(IdleAfter + 5*time.Second)
m.now = func() time.Time { return muchLater }
if got := m.State(); got != StateDegraded {
t.Fatalf("state after reconnect-but-no-events: %s, want %s", got, StateDegraded)
}
}
func TestMachine_ReconnectCount(t *testing.T) {
m := NewMachine()
m.OnConnected()
m.OnReconnect()
m.OnReconnect()
m.OnReconnect()
if snap := m.Snapshot(); snap.ReconnectCount != 3 {
t.Fatalf("ReconnectCount = %d, want 3", snap.ReconnectCount)
}
}
func TestMachine_OnEventClearsReconnecting(t *testing.T) {
m := NewMachine()
m.OnConnected()
m.OnReconnect()
if m.State() != StateReconnecting {
t.Fatal("expected StateReconnecting after OnReconnect")
}
m.OnEvent()
if m.State() != StateConnected {
t.Fatalf("OnEvent should clear Reconnecting, got %s", m.State())
}
}