初始化
This commit is contained in:
SunYF
@@ -1,54 +0,0 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
|
||||
"bj_power_mes/internal/eventbus"
|
||||
"bj_power_mes/internal/eventlog"
|
||||
)
|
||||
|
||||
// EventBridge 将 EventBus 事件转发给 SSE 客户端，实现内部事件到前端的实时推送。
|
||||
type EventBridge struct {
|
||||
sseHandler *Handler
|
||||
bus eventbus.Bus
|
||||
sub eventbus.Subscription
|
||||
}
|
||||
|
||||
// NewEventBridge 创建桥接器并订阅所有 EventBus 事件。
|
||||
func NewEventBridge(bus eventbus.Bus, sseHandler *Handler) *EventBridge {
|
||||
bridge := &EventBridge{
|
||||
sseHandler: sseHandler,
|
||||
bus: bus,
|
||||
}
|
||||
bridge.sub = bus.SubscribeAll(bridge.onEvent)
|
||||
return bridge
|
||||
}
|
||||
|
||||
// Close 取消事件订阅。
|
||||
func (b *EventBridge) Close() {
|
||||
if b.sub != nil {
|
||||
b.sub.Unsubscribe()
|
||||
}
|
||||
}
|
||||
|
||||
func (b *EventBridge) onEvent(_ context.Context, event eventbus.Event) error {
|
||||
data, err := json.Marshal(map[string]any{
|
||||
"id": event.ID,
|
||||
"eventType": string(event.Type),
|
||||
"sourceId": event.Source,
|
||||
"entityType": event.EntityType,
|
||||
"entityId": event.EntityID,
|
||||
"entityVersion": event.EntityVersion,
|
||||
"description": eventlog.EventDescription(event),
|
||||
"payload": event.Payload,
|
||||
"createdAt": event.Timestamp.Format("2006-01-02 15:04:05"),
|
||||
})
|
||||
if err != nil {
|
||||
slog.Error("eventbridge: marshal failed", "err", err)
|
||||
return nil
|
||||
}
|
||||
b.sseHandler.Emit("eventbus."+string(event.Type), string(data))
|
||||
return nil
|
||||
}
|
||||
@@ -1,79 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Beijing Hardman Automation Equipment Co., LTD. All rights reserved.
|
||||
*
|
||||
* Project: adapter
|
||||
* File: message.go
|
||||
* Last: 2025-04-15 16:59:49
|
||||
* Author: wangcheng@bj-hardman.com
|
||||
*/
|
||||
|
||||
package sse
|
||||
|
||||
// ToastType 'success' | 'warning' | 'error' | 'info' | 'default'
|
||||
type ToastType string
|
||||
|
||||
const (
|
||||
ToastTypeSuccess ToastType = "success"
|
||||
ToastTypeWarning ToastType = "warning"
|
||||
ToastTypeError ToastType = "error"
|
||||
ToastTypeInfo ToastType = "info"
|
||||
ToastTypeDefault ToastType = "default"
|
||||
)
|
||||
|
||||
type ToastMessage struct {
|
||||
Type ToastType `json:"type,omitempty"`
|
||||
Content string `json:"content"`
|
||||
Duration int `json:"duration,omitempty"` // 单位: 毫秒 默认:
|
||||
ShowClose bool `json:"showClose,omitempty"`
|
||||
}
|
||||
|
||||
type NoticeType string
|
||||
|
||||
const (
|
||||
NoticeTypeSuccess NoticeType = "success"
|
||||
NoticeTypeWarning NoticeType = "warning"
|
||||
NoticeTypeError NoticeType = "error"
|
||||
NoticeTypeInfo NoticeType = "info"
|
||||
NoticeTypeDefault NoticeType = "default"
|
||||
)
|
||||
|
||||
// NoticePosition 'top' | 'bottom' | 'topLeft' | 'topRight' | 'bottomLeft' | 'bottomRight'
|
||||
type NoticePosition string
|
||||
|
||||
const (
|
||||
NoticePositionTop NoticePosition = "top"
|
||||
NoticePositionBottom NoticePosition = "bottom"
|
||||
NoticePositionTopLeft NoticePosition = "topLeft"
|
||||
NoticePositionTopRight NoticePosition = "topRight"
|
||||
NoticePositionBottomLeft NoticePosition = "bottomLeft"
|
||||
NoticePositionBottomRight NoticePosition = "bottomRight"
|
||||
)
|
||||
|
||||
type NoticeMessage struct {
|
||||
Type NoticeType `json:"type,omitempty"`
|
||||
Title string `json:"title"`
|
||||
Content string `json:"content"`
|
||||
Duration int `json:"duration,omitempty"` // 单位: 毫秒 默认: 3000 -1表示无限长
|
||||
Position NoticePosition `json:"position,omitempty"`
|
||||
}
|
||||
|
||||
// Progress event types for real-time task progress推送
|
||||
const (
|
||||
EventTaskStarted = "task_started"
|
||||
EventTaskStepStarted = "task_step_started"
|
||||
EventTaskStepCompleted = "task_step_completed"
|
||||
EventTaskStepSkipped = "task_step_skipped"
|
||||
EventTaskStepFailed = "task_step_failed"
|
||||
EventTaskPaused = "task_paused"
|
||||
EventTaskResumed = "task_resumed"
|
||||
EventTaskError = "task_error"
|
||||
EventTaskFinished = "task_finished"
|
||||
EventTaskCancelled = "task_cancelled"
|
||||
EventMachineComplete = "machine_complete"
|
||||
EventWashComplete = "wash_complete"
|
||||
EventDeburComplete = "debur_complete"
|
||||
EventCleaningComplete = "cleaning_complete"
|
||||
EventTempStationUpdate = "temp_station_update"
|
||||
EventDockDataUpdate = "dock_data_update"
|
||||
EventDashboardUpdate = "dashboard_update" // 看板数据变更通知（Dashboard A 通过 SSE 订阅）
|
||||
)
|
||||
@@ -1,138 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Beijing Hardman Automation Equipment Co., LTD. All rights reserved.
|
||||
*
|
||||
* Project: adapter
|
||||
* File: sse.go
|
||||
* Last: 2025-05-07 14:12:54
|
||||
* Author: wangcheng@bj-hardman.com
|
||||
*/
|
||||
|
||||
package sse
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/zeromicro/go-zero/core/logx"
|
||||
)
|
||||
|
||||
type Handler struct {
|
||||
mu sync.RWMutex
|
||||
clients map[chan Event]bool
|
||||
}
|
||||
|
||||
type Event struct {
|
||||
Event string `json:"event"`
|
||||
Data string `json:"data"`
|
||||
}
|
||||
|
||||
func NewHandler() *Handler {
|
||||
return &Handler{
|
||||
clients: make(map[chan Event]bool),
|
||||
}
|
||||
}
|
||||
|
||||
// Serve 处理 SSE 连接
|
||||
func (h *Handler) Serve(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
|
||||
// 为每个客户端创建一个 channel
|
||||
clientChan := make(chan Event)
|
||||
h.mu.Lock()
|
||||
h.clients[clientChan] = true
|
||||
h.mu.Unlock()
|
||||
|
||||
// 客户端断开时清理
|
||||
defer func() {
|
||||
h.mu.Lock()
|
||||
delete(h.clients, clientChan)
|
||||
h.mu.Unlock()
|
||||
close(clientChan)
|
||||
}()
|
||||
|
||||
w.(http.Flusher).Flush() // 刷新缓冲区
|
||||
|
||||
heartbeat := time.NewTicker(30 * time.Second)
|
||||
defer heartbeat.Stop()
|
||||
|
||||
// 持续监听并推送事件
|
||||
for {
|
||||
select {
|
||||
case msg := <-clientChan:
|
||||
// 发送事件数据
|
||||
var err error
|
||||
if msg.Data != "" {
|
||||
_, err = fmt.Fprintf(w, "event: %s\ndata: %s\n\n", msg.Event, msg.Data)
|
||||
} else {
|
||||
_, err = fmt.Fprintf(w, "event: %s\n\n", msg.Event)
|
||||
}
|
||||
if err != nil {
|
||||
logx.Errorf("sse sent %+v failed: %s", msg, err)
|
||||
return
|
||||
}
|
||||
w.(http.Flusher).Flush()
|
||||
slog.InfoContext(r.Context(), fmt.Sprintf("sse sent %+v", msg))
|
||||
case <-heartbeat.C:
|
||||
_, err := fmt.Fprintf(w, ": heartbeat\n\n")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
w.(http.Flusher).Flush()
|
||||
case <-r.Context().Done():
|
||||
//客户端断开连接
|
||||
logx.Error(r.Context().Err())
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Handler) Emit(event string, data ...any) {
|
||||
// 广播给所有客户端
|
||||
|
||||
var js []byte
|
||||
if len(data) > 0 {
|
||||
if s, ok := data[0].(string); ok {
|
||||
js = []byte(s)
|
||||
} else {
|
||||
js, _ = json.Marshal(data[0])
|
||||
}
|
||||
}
|
||||
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
for msgChan := range h.clients {
|
||||
go func(ch chan Event) {
|
||||
defer func() { recover() }() // 客户端断连后 channel 已关闭，忽略 panic
|
||||
ch <- Event{
|
||||
Event: event,
|
||||
Data: string(js),
|
||||
}
|
||||
}(msgChan)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Handler) Toast(msg ToastMessage) {
|
||||
h.Emit("toast", msg)
|
||||
}
|
||||
|
||||
func (h *Handler) Notify(msg NoticeMessage) {
|
||||
h.Emit("notify", msg)
|
||||
}
|
||||
|
||||
// EmitProgressEvent sends a structured progress event to all SSE clients.
|
||||
func (h *Handler) EmitProgressEvent(eventType string, taskID int, stepName string, data map[string]any) {
|
||||
payload := map[string]any{
|
||||
"event": eventType,
|
||||
"task_id": taskID,
|
||||
"step_name": stepName,
|
||||
"data": data,
|
||||
}
|
||||
h.Emit("workflow", payload)
|
||||
}
|
||||
Reference in New Issue
Block a user