diff --git a/daemon/started_service.go b/daemon/started_service.go index 7176f058e..862b9920c 100644 --- a/daemon/started_service.go +++ b/daemon/started_service.go @@ -56,6 +56,9 @@ type StartedService struct { urlTestHistoryStorage *urltest.HistoryStorage clashModeSubscriber *observable.Subscriber[struct{}] clashModeObserver *observable.Observer[struct{}] + + connectionEventSubscriber *observable.Subscriber[trafficontrol.ConnectionEvent] + connectionEventObserver *observable.Observer[trafficontrol.ConnectionEvent] } type ServiceOptions struct { @@ -83,17 +86,19 @@ func NewStartedService(options ServiceOptions) *StartedService { // userID: options.UserID, // groupID: options.GroupID, // systemProxyEnabled: options.SystemProxyEnabled, - serviceStatus: &ServiceStatus{Status: ServiceStatus_IDLE}, - serviceStatusSubscriber: observable.NewSubscriber[*ServiceStatus](4), - logSubscriber: observable.NewSubscriber[*log.Entry](128), - urlTestSubscriber: observable.NewSubscriber[struct{}](1), - urlTestHistoryStorage: urltest.NewHistoryStorage(), - clashModeSubscriber: observable.NewSubscriber[struct{}](1), + serviceStatus: &ServiceStatus{Status: ServiceStatus_IDLE}, + serviceStatusSubscriber: observable.NewSubscriber[*ServiceStatus](4), + logSubscriber: observable.NewSubscriber[*log.Entry](128), + urlTestSubscriber: observable.NewSubscriber[struct{}](1), + urlTestHistoryStorage: urltest.NewHistoryStorage(), + clashModeSubscriber: observable.NewSubscriber[struct{}](1), + connectionEventSubscriber: observable.NewSubscriber[trafficontrol.ConnectionEvent](256), } s.serviceStatusObserver = observable.NewObserver(s.serviceStatusSubscriber, 2) s.logObserver = observable.NewObserver(s.logSubscriber, 64) s.urlTestObserver = observable.NewObserver(s.urlTestSubscriber, 1) s.clashModeObserver = observable.NewObserver(s.clashModeSubscriber, 1) + s.connectionEventObserver = observable.NewObserver(s.connectionEventSubscriber, 64) return s } @@ -183,6 +188,7 @@ func (s *StartedService) StartOrReloadService(profileContent string, options *Ov instance.urlTestHistoryStorage.SetHook(s.urlTestSubscriber) if instance.clashServer != nil { instance.clashServer.SetModeUpdateHook(s.clashModeSubscriber) + instance.clashServer.(*clashapi.Server).TrafficManager().SetEventHook(s.connectionEventSubscriber) } s.serviceAccess.Unlock() err = instance.Start() @@ -666,7 +672,7 @@ func (s *StartedService) SetSystemProxyEnabled(ctx context.Context, request *Set return nil, err } -func (s *StartedService) SubscribeConnections(request *SubscribeConnectionsRequest, server grpc.ServerStreamingServer[Connections]) error { +func (s *StartedService) SubscribeConnections(request *SubscribeConnectionsRequest, server grpc.ServerStreamingServer[ConnectionEvents]) error { err := s.waitForStarted(server.Context()) if err != nil { return err @@ -674,69 +680,253 @@ func (s *StartedService) SubscribeConnections(request *SubscribeConnectionsReque s.serviceAccess.RLock() boxService := s.instance s.serviceAccess.RUnlock() - ticker := time.NewTicker(time.Duration(request.Interval)) - defer ticker.Stop() + + if boxService.clashServer == nil { + return E.New("clash server not available") + } + trafficManager := boxService.clashServer.(*clashapi.Server).TrafficManager() - var ( - connections = make(map[uuid.UUID]*Connection) - outConnections []*Connection - ) + + subscription, done, err := s.connectionEventObserver.Subscribe() + if err != nil { + return err + } + defer s.connectionEventObserver.UnSubscribe(subscription) + + connectionSnapshots := make(map[uuid.UUID]connectionSnapshot) + initialEvents := s.buildInitialConnectionState(trafficManager, connectionSnapshots) + err = server.Send(&ConnectionEvents{ + Events: initialEvents, + Reset_: true, + }) + if err != nil { + return err + } + + interval := time.Duration(request.Interval) + if interval <= 0 { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { - outConnections = outConnections[:0] - for _, connection := range trafficManager.Connections() { - outConnections = append(outConnections, newConnection(connections, connection, false)) - } - for _, connection := range trafficManager.ClosedConnections() { - outConnections = append(outConnections, newConnection(connections, connection, true)) - } - err := server.Send(&Connections{Connections: outConnections}) - if err != nil { - return err - } select { case <-s.ctx.Done(): return s.ctx.Err() case <-server.Context().Done(): return server.Context().Err() + case <-done: + return nil + + case event := <-subscription: + var pendingEvents []*ConnectionEvent + if protoEvent := s.applyConnectionEvent(event, connectionSnapshots); protoEvent != nil { + pendingEvents = append(pendingEvents, protoEvent) + } + drain: + for { + select { + case event = <-subscription: + if protoEvent := s.applyConnectionEvent(event, connectionSnapshots); protoEvent != nil { + pendingEvents = append(pendingEvents, protoEvent) + } + default: + break drain + } + } + if len(pendingEvents) > 0 { + err = server.Send(&ConnectionEvents{Events: pendingEvents}) + if err != nil { + return err + } + } + case <-ticker.C: + protoEvents := s.buildTrafficUpdates(trafficManager, connectionSnapshots) + if len(protoEvents) == 0 { + continue + } + err = server.Send(&ConnectionEvents{Events: protoEvents}) + if err != nil { + return err + } } } } -func newConnection(connections map[uuid.UUID]*Connection, metadata trafficontrol.TrackerMetadata, isClosed bool) *Connection { - if oldConnection, loaded := connections[metadata.ID]; loaded { - if isClosed { - if oldConnection.ClosedAt == 0 { - oldConnection.Uplink = 0 - oldConnection.Downlink = 0 - oldConnection.ClosedAt = metadata.ClosedAt.UnixMilli() - } - return oldConnection +type connectionSnapshot struct { + uplink int64 + downlink int64 + hadTraffic bool +} + +func (s *StartedService) buildInitialConnectionState(manager *trafficontrol.Manager, snapshots map[uuid.UUID]connectionSnapshot) []*ConnectionEvent { + var events []*ConnectionEvent + + for _, metadata := range manager.Connections() { + events = append(events, &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_NEW, + Id: metadata.ID.String(), + Connection: buildConnectionProto(metadata), + }) + snapshots[metadata.ID] = connectionSnapshot{ + uplink: metadata.Upload.Load(), + downlink: metadata.Download.Load(), } - lastUplink := oldConnection.UplinkTotal - lastDownlink := oldConnection.DownlinkTotal - uplinkTotal := metadata.Upload.Load() - downlinkTotal := metadata.Download.Load() - oldConnection.Uplink = uplinkTotal - lastUplink - oldConnection.Downlink = downlinkTotal - lastDownlink - oldConnection.UplinkTotal = uplinkTotal - oldConnection.DownlinkTotal = downlinkTotal - return oldConnection } + + for _, metadata := range manager.ClosedConnections() { + conn := buildConnectionProto(metadata) + conn.ClosedAt = metadata.ClosedAt.UnixMilli() + events = append(events, &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_NEW, + Id: metadata.ID.String(), + Connection: conn, + }) + } + + return events +} + +func (s *StartedService) applyConnectionEvent(event trafficontrol.ConnectionEvent, snapshots map[uuid.UUID]connectionSnapshot) *ConnectionEvent { + switch event.Type { + case trafficontrol.ConnectionEventNew: + if _, exists := snapshots[event.ID]; exists { + return nil + } + snapshots[event.ID] = connectionSnapshot{ + uplink: event.Metadata.Upload.Load(), + downlink: event.Metadata.Download.Load(), + } + return &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_NEW, + Id: event.ID.String(), + Connection: buildConnectionProto(event.Metadata), + } + case trafficontrol.ConnectionEventClosed: + delete(snapshots, event.ID) + protoEvent := &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_CLOSED, + Id: event.ID.String(), + } + closedAt := event.ClosedAt + if closedAt.IsZero() && !event.Metadata.ClosedAt.IsZero() { + closedAt = event.Metadata.ClosedAt + } + if closedAt.IsZero() { + closedAt = time.Now() + } + protoEvent.ClosedAt = closedAt.UnixMilli() + if event.Metadata.ID != uuid.Nil { + conn := buildConnectionProto(event.Metadata) + conn.ClosedAt = protoEvent.ClosedAt + protoEvent.Connection = conn + } + return protoEvent + default: + return nil + } +} + +func (s *StartedService) buildTrafficUpdates(manager *trafficontrol.Manager, snapshots map[uuid.UUID]connectionSnapshot) []*ConnectionEvent { + activeConnections := manager.Connections() + activeIndex := make(map[uuid.UUID]trafficontrol.TrackerMetadata, len(activeConnections)) + var events []*ConnectionEvent + + for _, metadata := range activeConnections { + activeIndex[metadata.ID] = metadata + currentUpload := metadata.Upload.Load() + currentDownload := metadata.Download.Load() + snapshot, exists := snapshots[metadata.ID] + if !exists { + snapshots[metadata.ID] = connectionSnapshot{ + uplink: currentUpload, + downlink: currentDownload, + } + events = append(events, &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_NEW, + Id: metadata.ID.String(), + Connection: buildConnectionProto(metadata), + }) + continue + } + uplinkDelta := currentUpload - snapshot.uplink + downlinkDelta := currentDownload - snapshot.downlink + if uplinkDelta < 0 || downlinkDelta < 0 { + snapshots[metadata.ID] = connectionSnapshot{ + uplink: currentUpload, + downlink: currentDownload, + } + continue + } + if uplinkDelta > 0 || downlinkDelta > 0 { + snapshots[metadata.ID] = connectionSnapshot{ + uplink: currentUpload, + downlink: currentDownload, + hadTraffic: true, + } + events = append(events, &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_UPDATE, + Id: metadata.ID.String(), + UplinkDelta: uplinkDelta, + DownlinkDelta: downlinkDelta, + }) + continue + } + if snapshot.hadTraffic { + snapshots[metadata.ID] = connectionSnapshot{ + uplink: currentUpload, + downlink: currentDownload, + } + events = append(events, &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_UPDATE, + Id: metadata.ID.String(), + UplinkDelta: 0, + DownlinkDelta: 0, + }) + } + } + + var closedIndex map[uuid.UUID]trafficontrol.TrackerMetadata + for id := range snapshots { + if _, exists := activeIndex[id]; exists { + continue + } + if closedIndex == nil { + closedIndex = make(map[uuid.UUID]trafficontrol.TrackerMetadata) + for _, metadata := range manager.ClosedConnections() { + closedIndex[metadata.ID] = metadata + } + } + closedAt := time.Now() + var conn *Connection + if metadata, ok := closedIndex[id]; ok { + if !metadata.ClosedAt.IsZero() { + closedAt = metadata.ClosedAt + } + conn = buildConnectionProto(metadata) + conn.ClosedAt = closedAt.UnixMilli() + } + events = append(events, &ConnectionEvent{ + Type: ConnectionEventType_CONNECTION_EVENT_CLOSED, + Id: id.String(), + ClosedAt: closedAt.UnixMilli(), + Connection: conn, + }) + delete(snapshots, id) + } + + return events +} + +func buildConnectionProto(metadata trafficontrol.TrackerMetadata) *Connection { var rule string if metadata.Rule != nil { rule = metadata.Rule.String() } uplinkTotal := metadata.Upload.Load() downlinkTotal := metadata.Download.Load() - uplink := uplinkTotal - downlink := downlinkTotal - var closedAt int64 - if !metadata.ClosedAt.IsZero() { - closedAt = metadata.ClosedAt.UnixMilli() - uplink = 0 - downlink = 0 - } var processInfo *ProcessInfo if metadata.Metadata.ProcessInfo != nil { processInfo = &ProcessInfo{ @@ -747,7 +937,7 @@ func newConnection(connections map[uuid.UUID]*Connection, metadata trafficontrol PackageName: metadata.Metadata.ProcessInfo.AndroidPackageName, } } - connection := &Connection{ + return &Connection{ Id: metadata.ID.String(), Inbound: metadata.Metadata.Inbound, InboundType: metadata.Metadata.InboundType, @@ -760,9 +950,6 @@ func newConnection(connections map[uuid.UUID]*Connection, metadata trafficontrol User: metadata.Metadata.User, FromOutbound: metadata.Metadata.Outbound, CreatedAt: metadata.CreatedAt.UnixMilli(), - ClosedAt: closedAt, - Uplink: uplink, - Downlink: downlink, UplinkTotal: uplinkTotal, DownlinkTotal: downlinkTotal, Rule: rule, @@ -771,8 +958,6 @@ func newConnection(connections map[uuid.UUID]*Connection, metadata trafficontrol ChainList: metadata.Chain, ProcessInfo: processInfo, } - connections[metadata.ID] = connection - return connection } func (s *StartedService) CloseConnection(ctx context.Context, request *CloseConnectionRequest) (*emptypb.Empty, error) { diff --git a/daemon/started_service.pb.go b/daemon/started_service.pb.go index b00a1fb2e..ef9ea8250 100644 --- a/daemon/started_service.pb.go +++ b/daemon/started_service.pb.go @@ -78,104 +78,55 @@ func (LogLevel) EnumDescriptor() ([]byte, []int) { return file_daemon_started_service_proto_rawDescGZIP(), []int{0} } -type ConnectionFilter int32 +type ConnectionEventType int32 const ( - ConnectionFilter_ALL ConnectionFilter = 0 - ConnectionFilter_ACTIVE ConnectionFilter = 1 - ConnectionFilter_CLOSED ConnectionFilter = 2 + ConnectionEventType_CONNECTION_EVENT_NEW ConnectionEventType = 0 + ConnectionEventType_CONNECTION_EVENT_UPDATE ConnectionEventType = 1 + ConnectionEventType_CONNECTION_EVENT_CLOSED ConnectionEventType = 2 ) -// Enum value maps for ConnectionFilter. +// Enum value maps for ConnectionEventType. var ( - ConnectionFilter_name = map[int32]string{ - 0: "ALL", - 1: "ACTIVE", - 2: "CLOSED", + ConnectionEventType_name = map[int32]string{ + 0: "CONNECTION_EVENT_NEW", + 1: "CONNECTION_EVENT_UPDATE", + 2: "CONNECTION_EVENT_CLOSED", } - ConnectionFilter_value = map[string]int32{ - "ALL": 0, - "ACTIVE": 1, - "CLOSED": 2, + ConnectionEventType_value = map[string]int32{ + "CONNECTION_EVENT_NEW": 0, + "CONNECTION_EVENT_UPDATE": 1, + "CONNECTION_EVENT_CLOSED": 2, } ) -func (x ConnectionFilter) Enum() *ConnectionFilter { - p := new(ConnectionFilter) +func (x ConnectionEventType) Enum() *ConnectionEventType { + p := new(ConnectionEventType) *p = x return p } -func (x ConnectionFilter) String() string { +func (x ConnectionEventType) String() string { return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) } -func (ConnectionFilter) Descriptor() protoreflect.EnumDescriptor { +func (ConnectionEventType) Descriptor() protoreflect.EnumDescriptor { return file_daemon_started_service_proto_enumTypes[1].Descriptor() } -func (ConnectionFilter) Type() protoreflect.EnumType { +func (ConnectionEventType) Type() protoreflect.EnumType { return &file_daemon_started_service_proto_enumTypes[1] } -func (x ConnectionFilter) Number() protoreflect.EnumNumber { +func (x ConnectionEventType) Number() protoreflect.EnumNumber { return protoreflect.EnumNumber(x) } -// Deprecated: Use ConnectionFilter.Descriptor instead. -func (ConnectionFilter) EnumDescriptor() ([]byte, []int) { +// Deprecated: Use ConnectionEventType.Descriptor instead. +func (ConnectionEventType) EnumDescriptor() ([]byte, []int) { return file_daemon_started_service_proto_rawDescGZIP(), []int{1} } -type ConnectionSortBy int32 - -const ( - ConnectionSortBy_DATE ConnectionSortBy = 0 - ConnectionSortBy_TRAFFIC ConnectionSortBy = 1 - ConnectionSortBy_TOTAL_TRAFFIC ConnectionSortBy = 2 -) - -// Enum value maps for ConnectionSortBy. -var ( - ConnectionSortBy_name = map[int32]string{ - 0: "DATE", - 1: "TRAFFIC", - 2: "TOTAL_TRAFFIC", - } - ConnectionSortBy_value = map[string]int32{ - "DATE": 0, - "TRAFFIC": 1, - "TOTAL_TRAFFIC": 2, - } -) - -func (x ConnectionSortBy) Enum() *ConnectionSortBy { - p := new(ConnectionSortBy) - *p = x - return p -} - -func (x ConnectionSortBy) String() string { - return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) -} - -func (ConnectionSortBy) Descriptor() protoreflect.EnumDescriptor { - return file_daemon_started_service_proto_enumTypes[2].Descriptor() -} - -func (ConnectionSortBy) Type() protoreflect.EnumType { - return &file_daemon_started_service_proto_enumTypes[2] -} - -func (x ConnectionSortBy) Number() protoreflect.EnumNumber { - return protoreflect.EnumNumber(x) -} - -// Deprecated: Use ConnectionSortBy.Descriptor instead. -func (ConnectionSortBy) EnumDescriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{2} -} - type ServiceStatus_Type int32 const ( @@ -215,11 +166,11 @@ func (x ServiceStatus_Type) String() string { } func (ServiceStatus_Type) Descriptor() protoreflect.EnumDescriptor { - return file_daemon_started_service_proto_enumTypes[3].Descriptor() + return file_daemon_started_service_proto_enumTypes[2].Descriptor() } func (ServiceStatus_Type) Type() protoreflect.EnumType { - return &file_daemon_started_service_proto_enumTypes[3] + return &file_daemon_started_service_proto_enumTypes[2] } func (x ServiceStatus_Type) Number() protoreflect.EnumNumber { @@ -1114,8 +1065,6 @@ func (x *SetSystemProxyEnabledRequest) GetEnabled() bool { type SubscribeConnectionsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Interval int64 `protobuf:"varint,1,opt,name=interval,proto3" json:"interval,omitempty"` - Filter ConnectionFilter `protobuf:"varint,2,opt,name=filter,proto3,enum=daemon.ConnectionFilter" json:"filter,omitempty"` - SortBy ConnectionSortBy `protobuf:"varint,3,opt,name=sortBy,proto3,enum=daemon.ConnectionSortBy" json:"sortBy,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1157,41 +1106,32 @@ func (x *SubscribeConnectionsRequest) GetInterval() int64 { return 0 } -func (x *SubscribeConnectionsRequest) GetFilter() ConnectionFilter { - if x != nil { - return x.Filter - } - return ConnectionFilter_ALL -} - -func (x *SubscribeConnectionsRequest) GetSortBy() ConnectionSortBy { - if x != nil { - return x.SortBy - } - return ConnectionSortBy_DATE -} - -type Connections struct { +type ConnectionEvent struct { state protoimpl.MessageState `protogen:"open.v1"` - Connections []*Connection `protobuf:"bytes,1,rep,name=connections,proto3" json:"connections,omitempty"` + Type ConnectionEventType `protobuf:"varint,1,opt,name=type,proto3,enum=daemon.ConnectionEventType" json:"type,omitempty"` + Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"` + Connection *Connection `protobuf:"bytes,3,opt,name=connection,proto3" json:"connection,omitempty"` + UplinkDelta int64 `protobuf:"varint,4,opt,name=uplinkDelta,proto3" json:"uplinkDelta,omitempty"` + DownlinkDelta int64 `protobuf:"varint,5,opt,name=downlinkDelta,proto3" json:"downlinkDelta,omitempty"` + ClosedAt int64 `protobuf:"varint,6,opt,name=closedAt,proto3" json:"closedAt,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } -func (x *Connections) Reset() { - *x = Connections{} +func (x *ConnectionEvent) Reset() { + *x = ConnectionEvent{} mi := &file_daemon_started_service_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } -func (x *Connections) String() string { +func (x *ConnectionEvent) String() string { return protoimpl.X.MessageStringOf(x) } -func (*Connections) ProtoMessage() {} +func (*ConnectionEvent) ProtoMessage() {} -func (x *Connections) ProtoReflect() protoreflect.Message { +func (x *ConnectionEvent) ProtoReflect() protoreflect.Message { mi := &file_daemon_started_service_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -1203,18 +1143,105 @@ func (x *Connections) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use Connections.ProtoReflect.Descriptor instead. -func (*Connections) Descriptor() ([]byte, []int) { +// Deprecated: Use ConnectionEvent.ProtoReflect.Descriptor instead. +func (*ConnectionEvent) Descriptor() ([]byte, []int) { return file_daemon_started_service_proto_rawDescGZIP(), []int{17} } -func (x *Connections) GetConnections() []*Connection { +func (x *ConnectionEvent) GetType() ConnectionEventType { if x != nil { - return x.Connections + return x.Type + } + return ConnectionEventType_CONNECTION_EVENT_NEW +} + +func (x *ConnectionEvent) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *ConnectionEvent) GetConnection() *Connection { + if x != nil { + return x.Connection } return nil } +func (x *ConnectionEvent) GetUplinkDelta() int64 { + if x != nil { + return x.UplinkDelta + } + return 0 +} + +func (x *ConnectionEvent) GetDownlinkDelta() int64 { + if x != nil { + return x.DownlinkDelta + } + return 0 +} + +func (x *ConnectionEvent) GetClosedAt() int64 { + if x != nil { + return x.ClosedAt + } + return 0 +} + +type ConnectionEvents struct { + state protoimpl.MessageState `protogen:"open.v1"` + Events []*ConnectionEvent `protobuf:"bytes,1,rep,name=events,proto3" json:"events,omitempty"` + Reset_ bool `protobuf:"varint,2,opt,name=reset,proto3" json:"reset,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ConnectionEvents) Reset() { + *x = ConnectionEvents{} + mi := &file_daemon_started_service_proto_msgTypes[18] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ConnectionEvents) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ConnectionEvents) ProtoMessage() {} + +func (x *ConnectionEvents) ProtoReflect() protoreflect.Message { + mi := &file_daemon_started_service_proto_msgTypes[18] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ConnectionEvents.ProtoReflect.Descriptor instead. +func (*ConnectionEvents) Descriptor() ([]byte, []int) { + return file_daemon_started_service_proto_rawDescGZIP(), []int{18} +} + +func (x *ConnectionEvents) GetEvents() []*ConnectionEvent { + if x != nil { + return x.Events + } + return nil +} + +func (x *ConnectionEvents) GetReset_() bool { + if x != nil { + return x.Reset_ + } + return false +} + type Connection struct { state protoimpl.MessageState `protogen:"open.v1"` Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` @@ -1245,7 +1272,7 @@ type Connection struct { func (x *Connection) Reset() { *x = Connection{} - mi := &file_daemon_started_service_proto_msgTypes[18] + mi := &file_daemon_started_service_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1257,7 +1284,7 @@ func (x *Connection) String() string { func (*Connection) ProtoMessage() {} func (x *Connection) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[18] + mi := &file_daemon_started_service_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1270,7 +1297,7 @@ func (x *Connection) ProtoReflect() protoreflect.Message { // Deprecated: Use Connection.ProtoReflect.Descriptor instead. func (*Connection) Descriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{18} + return file_daemon_started_service_proto_rawDescGZIP(), []int{19} } func (x *Connection) GetId() string { @@ -1440,7 +1467,7 @@ type ProcessInfo struct { func (x *ProcessInfo) Reset() { *x = ProcessInfo{} - mi := &file_daemon_started_service_proto_msgTypes[19] + mi := &file_daemon_started_service_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1452,7 +1479,7 @@ func (x *ProcessInfo) String() string { func (*ProcessInfo) ProtoMessage() {} func (x *ProcessInfo) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[19] + mi := &file_daemon_started_service_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1465,7 +1492,7 @@ func (x *ProcessInfo) ProtoReflect() protoreflect.Message { // Deprecated: Use ProcessInfo.ProtoReflect.Descriptor instead. func (*ProcessInfo) Descriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{19} + return file_daemon_started_service_proto_rawDescGZIP(), []int{20} } func (x *ProcessInfo) GetProcessId() uint32 { @@ -1512,7 +1539,7 @@ type CloseConnectionRequest struct { func (x *CloseConnectionRequest) Reset() { *x = CloseConnectionRequest{} - mi := &file_daemon_started_service_proto_msgTypes[20] + mi := &file_daemon_started_service_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1524,7 +1551,7 @@ func (x *CloseConnectionRequest) String() string { func (*CloseConnectionRequest) ProtoMessage() {} func (x *CloseConnectionRequest) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[20] + mi := &file_daemon_started_service_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1537,7 +1564,7 @@ func (x *CloseConnectionRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CloseConnectionRequest.ProtoReflect.Descriptor instead. func (*CloseConnectionRequest) Descriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{20} + return file_daemon_started_service_proto_rawDescGZIP(), []int{21} } func (x *CloseConnectionRequest) GetId() string { @@ -1556,7 +1583,7 @@ type DeprecatedWarnings struct { func (x *DeprecatedWarnings) Reset() { *x = DeprecatedWarnings{} - mi := &file_daemon_started_service_proto_msgTypes[21] + mi := &file_daemon_started_service_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1568,7 +1595,7 @@ func (x *DeprecatedWarnings) String() string { func (*DeprecatedWarnings) ProtoMessage() {} func (x *DeprecatedWarnings) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[21] + mi := &file_daemon_started_service_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1581,7 +1608,7 @@ func (x *DeprecatedWarnings) ProtoReflect() protoreflect.Message { // Deprecated: Use DeprecatedWarnings.ProtoReflect.Descriptor instead. func (*DeprecatedWarnings) Descriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{21} + return file_daemon_started_service_proto_rawDescGZIP(), []int{22} } func (x *DeprecatedWarnings) GetWarnings() []*DeprecatedWarning { @@ -1602,7 +1629,7 @@ type DeprecatedWarning struct { func (x *DeprecatedWarning) Reset() { *x = DeprecatedWarning{} - mi := &file_daemon_started_service_proto_msgTypes[22] + mi := &file_daemon_started_service_proto_msgTypes[23] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1614,7 +1641,7 @@ func (x *DeprecatedWarning) String() string { func (*DeprecatedWarning) ProtoMessage() {} func (x *DeprecatedWarning) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[22] + mi := &file_daemon_started_service_proto_msgTypes[23] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1627,7 +1654,7 @@ func (x *DeprecatedWarning) ProtoReflect() protoreflect.Message { // Deprecated: Use DeprecatedWarning.ProtoReflect.Descriptor instead. func (*DeprecatedWarning) Descriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{22} + return file_daemon_started_service_proto_rawDescGZIP(), []int{23} } func (x *DeprecatedWarning) GetMessage() string { @@ -1660,7 +1687,7 @@ type StartedAt struct { func (x *StartedAt) Reset() { *x = StartedAt{} - mi := &file_daemon_started_service_proto_msgTypes[23] + mi := &file_daemon_started_service_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1672,7 +1699,7 @@ func (x *StartedAt) String() string { func (*StartedAt) ProtoMessage() {} func (x *StartedAt) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[23] + mi := &file_daemon_started_service_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1685,7 +1712,7 @@ func (x *StartedAt) ProtoReflect() protoreflect.Message { // Deprecated: Use StartedAt.ProtoReflect.Descriptor instead. func (*StartedAt) Descriptor() ([]byte, []int) { - return file_daemon_started_service_proto_rawDescGZIP(), []int{23} + return file_daemon_started_service_proto_rawDescGZIP(), []int{24} } func (x *StartedAt) GetStartedAt() int64 { @@ -1705,7 +1732,7 @@ type Log_Message struct { func (x *Log_Message) Reset() { *x = Log_Message{} - mi := &file_daemon_started_service_proto_msgTypes[24] + mi := &file_daemon_started_service_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1717,7 +1744,7 @@ func (x *Log_Message) String() string { func (*Log_Message) ProtoMessage() {} func (x *Log_Message) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[24] + mi := &file_daemon_started_service_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1818,13 +1845,21 @@ const file_daemon_started_service_proto_rawDesc = "" + "\tavailable\x18\x01 \x01(\bR\tavailable\x12\x18\n" + "\aenabled\x18\x02 \x01(\bR\aenabled\"8\n" + "\x1cSetSystemProxyEnabledRequest\x12\x18\n" + - "\aenabled\x18\x01 \x01(\bR\aenabled\"\x9d\x01\n" + + "\aenabled\x18\x01 \x01(\bR\aenabled\"9\n" + "\x1bSubscribeConnectionsRequest\x12\x1a\n" + - "\binterval\x18\x01 \x01(\x03R\binterval\x120\n" + - "\x06filter\x18\x02 \x01(\x0e2\x18.daemon.ConnectionFilterR\x06filter\x120\n" + - "\x06sortBy\x18\x03 \x01(\x0e2\x18.daemon.ConnectionSortByR\x06sortBy\"C\n" + - "\vConnections\x124\n" + - "\vconnections\x18\x01 \x03(\v2\x12.daemon.ConnectionR\vconnections\"\x95\x05\n" + + "\binterval\x18\x01 \x01(\x03R\binterval\"\xea\x01\n" + + "\x0fConnectionEvent\x12/\n" + + "\x04type\x18\x01 \x01(\x0e2\x1b.daemon.ConnectionEventTypeR\x04type\x12\x0e\n" + + "\x02id\x18\x02 \x01(\tR\x02id\x122\n" + + "\n" + + "connection\x18\x03 \x01(\v2\x12.daemon.ConnectionR\n" + + "connection\x12 \n" + + "\vuplinkDelta\x18\x04 \x01(\x03R\vuplinkDelta\x12$\n" + + "\rdownlinkDelta\x18\x05 \x01(\x03R\rdownlinkDelta\x12\x1a\n" + + "\bclosedAt\x18\x06 \x01(\x03R\bclosedAt\"Y\n" + + "\x10ConnectionEvents\x12/\n" + + "\x06events\x18\x01 \x03(\v2\x17.daemon.ConnectionEventR\x06events\x12\x14\n" + + "\x05reset\x18\x02 \x01(\bR\x05reset\"\x95\x05\n" + "\n" + "Connection\x12\x0e\n" + "\x02id\x18\x01 \x01(\tR\x02id\x12\x18\n" + @@ -1873,17 +1908,11 @@ const file_daemon_started_service_proto_rawDesc = "" + "\x04WARN\x10\x03\x12\b\n" + "\x04INFO\x10\x04\x12\t\n" + "\x05DEBUG\x10\x05\x12\t\n" + - "\x05TRACE\x10\x06*3\n" + - "\x10ConnectionFilter\x12\a\n" + - "\x03ALL\x10\x00\x12\n" + - "\n" + - "\x06ACTIVE\x10\x01\x12\n" + - "\n" + - "\x06CLOSED\x10\x02*<\n" + - "\x10ConnectionSortBy\x12\b\n" + - "\x04DATE\x10\x00\x12\v\n" + - "\aTRAFFIC\x10\x01\x12\x11\n" + - "\rTOTAL_TRAFFIC\x10\x022\xe0\v\n" + + "\x05TRACE\x10\x06*i\n" + + "\x13ConnectionEventType\x12\x18\n" + + "\x14CONNECTION_EVENT_NEW\x10\x00\x12\x1b\n" + + "\x17CONNECTION_EVENT_UPDATE\x10\x01\x12\x1b\n" + + "\x17CONNECTION_EVENT_CLOSED\x10\x022\xe5\v\n" + "\x0eStartedService\x12=\n" + "\vStopService\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\x12?\n" + "\rReloadService\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\x12K\n" + @@ -1900,8 +1929,8 @@ const file_daemon_started_service_proto_rawDesc = "" + "\x0eSelectOutbound\x12\x1d.daemon.SelectOutboundRequest\x1a\x16.google.protobuf.Empty\"\x00\x12I\n" + "\x0eSetGroupExpand\x12\x1d.daemon.SetGroupExpandRequest\x1a\x16.google.protobuf.Empty\"\x00\x12K\n" + "\x14GetSystemProxyStatus\x12\x16.google.protobuf.Empty\x1a\x19.daemon.SystemProxyStatus\"\x00\x12W\n" + - "\x15SetSystemProxyEnabled\x12$.daemon.SetSystemProxyEnabledRequest\x1a\x16.google.protobuf.Empty\"\x00\x12T\n" + - "\x14SubscribeConnections\x12#.daemon.SubscribeConnectionsRequest\x1a\x13.daemon.Connections\"\x000\x01\x12K\n" + + "\x15SetSystemProxyEnabled\x12$.daemon.SetSystemProxyEnabledRequest\x1a\x16.google.protobuf.Empty\"\x00\x12Y\n" + + "\x14SubscribeConnections\x12#.daemon.SubscribeConnectionsRequest\x1a\x18.daemon.ConnectionEvents\"\x000\x01\x12K\n" + "\x0fCloseConnection\x12\x1e.daemon.CloseConnectionRequest\x1a\x16.google.protobuf.Empty\"\x00\x12G\n" + "\x13CloseAllConnections\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\"\x00\x12M\n" + "\x15GetDeprecatedWarnings\x12\x16.google.protobuf.Empty\x1a\x1a.daemon.DeprecatedWarnings\"\x00\x12;\n" + @@ -1920,31 +1949,31 @@ func file_daemon_started_service_proto_rawDescGZIP() []byte { } var ( - file_daemon_started_service_proto_enumTypes = make([]protoimpl.EnumInfo, 4) - file_daemon_started_service_proto_msgTypes = make([]protoimpl.MessageInfo, 25) + file_daemon_started_service_proto_enumTypes = make([]protoimpl.EnumInfo, 3) + file_daemon_started_service_proto_msgTypes = make([]protoimpl.MessageInfo, 26) file_daemon_started_service_proto_goTypes = []any{ (LogLevel)(0), // 0: daemon.LogLevel - (ConnectionFilter)(0), // 1: daemon.ConnectionFilter - (ConnectionSortBy)(0), // 2: daemon.ConnectionSortBy - (ServiceStatus_Type)(0), // 3: daemon.ServiceStatus.Type - (*ServiceStatus)(nil), // 4: daemon.ServiceStatus - (*ReloadServiceRequest)(nil), // 5: daemon.ReloadServiceRequest - (*SubscribeStatusRequest)(nil), // 6: daemon.SubscribeStatusRequest - (*Log)(nil), // 7: daemon.Log - (*DefaultLogLevel)(nil), // 8: daemon.DefaultLogLevel - (*Status)(nil), // 9: daemon.Status - (*Groups)(nil), // 10: daemon.Groups - (*Group)(nil), // 11: daemon.Group - (*GroupItem)(nil), // 12: daemon.GroupItem - (*URLTestRequest)(nil), // 13: daemon.URLTestRequest - (*SelectOutboundRequest)(nil), // 14: daemon.SelectOutboundRequest - (*SetGroupExpandRequest)(nil), // 15: daemon.SetGroupExpandRequest - (*ClashMode)(nil), // 16: daemon.ClashMode - (*ClashModeStatus)(nil), // 17: daemon.ClashModeStatus - (*SystemProxyStatus)(nil), // 18: daemon.SystemProxyStatus - (*SetSystemProxyEnabledRequest)(nil), // 19: daemon.SetSystemProxyEnabledRequest - (*SubscribeConnectionsRequest)(nil), // 20: daemon.SubscribeConnectionsRequest - (*Connections)(nil), // 21: daemon.Connections + (ConnectionEventType)(0), // 1: daemon.ConnectionEventType + (ServiceStatus_Type)(0), // 2: daemon.ServiceStatus.Type + (*ServiceStatus)(nil), // 3: daemon.ServiceStatus + (*ReloadServiceRequest)(nil), // 4: daemon.ReloadServiceRequest + (*SubscribeStatusRequest)(nil), // 5: daemon.SubscribeStatusRequest + (*Log)(nil), // 6: daemon.Log + (*DefaultLogLevel)(nil), // 7: daemon.DefaultLogLevel + (*Status)(nil), // 8: daemon.Status + (*Groups)(nil), // 9: daemon.Groups + (*Group)(nil), // 10: daemon.Group + (*GroupItem)(nil), // 11: daemon.GroupItem + (*URLTestRequest)(nil), // 12: daemon.URLTestRequest + (*SelectOutboundRequest)(nil), // 13: daemon.SelectOutboundRequest + (*SetGroupExpandRequest)(nil), // 14: daemon.SetGroupExpandRequest + (*ClashMode)(nil), // 15: daemon.ClashMode + (*ClashModeStatus)(nil), // 16: daemon.ClashModeStatus + (*SystemProxyStatus)(nil), // 17: daemon.SystemProxyStatus + (*SetSystemProxyEnabledRequest)(nil), // 18: daemon.SetSystemProxyEnabledRequest + (*SubscribeConnectionsRequest)(nil), // 19: daemon.SubscribeConnectionsRequest + (*ConnectionEvent)(nil), // 20: daemon.ConnectionEvent + (*ConnectionEvents)(nil), // 21: daemon.ConnectionEvents (*Connection)(nil), // 22: daemon.Connection (*ProcessInfo)(nil), // 23: daemon.ProcessInfo (*CloseConnectionRequest)(nil), // 24: daemon.CloseConnectionRequest @@ -1957,14 +1986,14 @@ var ( ) var file_daemon_started_service_proto_depIdxs = []int32{ - 3, // 0: daemon.ServiceStatus.status:type_name -> daemon.ServiceStatus.Type + 2, // 0: daemon.ServiceStatus.status:type_name -> daemon.ServiceStatus.Type 28, // 1: daemon.Log.messages:type_name -> daemon.Log.Message 0, // 2: daemon.DefaultLogLevel.level:type_name -> daemon.LogLevel - 11, // 3: daemon.Groups.group:type_name -> daemon.Group - 12, // 4: daemon.Group.items:type_name -> daemon.GroupItem - 1, // 5: daemon.SubscribeConnectionsRequest.filter:type_name -> daemon.ConnectionFilter - 2, // 6: daemon.SubscribeConnectionsRequest.sortBy:type_name -> daemon.ConnectionSortBy - 22, // 7: daemon.Connections.connections:type_name -> daemon.Connection + 10, // 3: daemon.Groups.group:type_name -> daemon.Group + 11, // 4: daemon.Group.items:type_name -> daemon.GroupItem + 1, // 5: daemon.ConnectionEvent.type:type_name -> daemon.ConnectionEventType + 22, // 6: daemon.ConnectionEvent.connection:type_name -> daemon.Connection + 20, // 7: daemon.ConnectionEvents.events:type_name -> daemon.ConnectionEvent 23, // 8: daemon.Connection.processInfo:type_name -> daemon.ProcessInfo 26, // 9: daemon.DeprecatedWarnings.warnings:type_name -> daemon.DeprecatedWarning 0, // 10: daemon.Log.Message.level:type_name -> daemon.LogLevel @@ -1974,38 +2003,38 @@ var file_daemon_started_service_proto_depIdxs = []int32{ 29, // 14: daemon.StartedService.SubscribeLog:input_type -> google.protobuf.Empty 29, // 15: daemon.StartedService.GetDefaultLogLevel:input_type -> google.protobuf.Empty 29, // 16: daemon.StartedService.ClearLogs:input_type -> google.protobuf.Empty - 6, // 17: daemon.StartedService.SubscribeStatus:input_type -> daemon.SubscribeStatusRequest + 5, // 17: daemon.StartedService.SubscribeStatus:input_type -> daemon.SubscribeStatusRequest 29, // 18: daemon.StartedService.SubscribeGroups:input_type -> google.protobuf.Empty 29, // 19: daemon.StartedService.GetClashModeStatus:input_type -> google.protobuf.Empty 29, // 20: daemon.StartedService.SubscribeClashMode:input_type -> google.protobuf.Empty - 16, // 21: daemon.StartedService.SetClashMode:input_type -> daemon.ClashMode - 13, // 22: daemon.StartedService.URLTest:input_type -> daemon.URLTestRequest - 14, // 23: daemon.StartedService.SelectOutbound:input_type -> daemon.SelectOutboundRequest - 15, // 24: daemon.StartedService.SetGroupExpand:input_type -> daemon.SetGroupExpandRequest + 15, // 21: daemon.StartedService.SetClashMode:input_type -> daemon.ClashMode + 12, // 22: daemon.StartedService.URLTest:input_type -> daemon.URLTestRequest + 13, // 23: daemon.StartedService.SelectOutbound:input_type -> daemon.SelectOutboundRequest + 14, // 24: daemon.StartedService.SetGroupExpand:input_type -> daemon.SetGroupExpandRequest 29, // 25: daemon.StartedService.GetSystemProxyStatus:input_type -> google.protobuf.Empty - 19, // 26: daemon.StartedService.SetSystemProxyEnabled:input_type -> daemon.SetSystemProxyEnabledRequest - 20, // 27: daemon.StartedService.SubscribeConnections:input_type -> daemon.SubscribeConnectionsRequest + 18, // 26: daemon.StartedService.SetSystemProxyEnabled:input_type -> daemon.SetSystemProxyEnabledRequest + 19, // 27: daemon.StartedService.SubscribeConnections:input_type -> daemon.SubscribeConnectionsRequest 24, // 28: daemon.StartedService.CloseConnection:input_type -> daemon.CloseConnectionRequest 29, // 29: daemon.StartedService.CloseAllConnections:input_type -> google.protobuf.Empty 29, // 30: daemon.StartedService.GetDeprecatedWarnings:input_type -> google.protobuf.Empty 29, // 31: daemon.StartedService.GetStartedAt:input_type -> google.protobuf.Empty 29, // 32: daemon.StartedService.StopService:output_type -> google.protobuf.Empty 29, // 33: daemon.StartedService.ReloadService:output_type -> google.protobuf.Empty - 4, // 34: daemon.StartedService.SubscribeServiceStatus:output_type -> daemon.ServiceStatus - 7, // 35: daemon.StartedService.SubscribeLog:output_type -> daemon.Log - 8, // 36: daemon.StartedService.GetDefaultLogLevel:output_type -> daemon.DefaultLogLevel + 3, // 34: daemon.StartedService.SubscribeServiceStatus:output_type -> daemon.ServiceStatus + 6, // 35: daemon.StartedService.SubscribeLog:output_type -> daemon.Log + 7, // 36: daemon.StartedService.GetDefaultLogLevel:output_type -> daemon.DefaultLogLevel 29, // 37: daemon.StartedService.ClearLogs:output_type -> google.protobuf.Empty - 9, // 38: daemon.StartedService.SubscribeStatus:output_type -> daemon.Status - 10, // 39: daemon.StartedService.SubscribeGroups:output_type -> daemon.Groups - 17, // 40: daemon.StartedService.GetClashModeStatus:output_type -> daemon.ClashModeStatus - 16, // 41: daemon.StartedService.SubscribeClashMode:output_type -> daemon.ClashMode + 8, // 38: daemon.StartedService.SubscribeStatus:output_type -> daemon.Status + 9, // 39: daemon.StartedService.SubscribeGroups:output_type -> daemon.Groups + 16, // 40: daemon.StartedService.GetClashModeStatus:output_type -> daemon.ClashModeStatus + 15, // 41: daemon.StartedService.SubscribeClashMode:output_type -> daemon.ClashMode 29, // 42: daemon.StartedService.SetClashMode:output_type -> google.protobuf.Empty 29, // 43: daemon.StartedService.URLTest:output_type -> google.protobuf.Empty 29, // 44: daemon.StartedService.SelectOutbound:output_type -> google.protobuf.Empty 29, // 45: daemon.StartedService.SetGroupExpand:output_type -> google.protobuf.Empty - 18, // 46: daemon.StartedService.GetSystemProxyStatus:output_type -> daemon.SystemProxyStatus + 17, // 46: daemon.StartedService.GetSystemProxyStatus:output_type -> daemon.SystemProxyStatus 29, // 47: daemon.StartedService.SetSystemProxyEnabled:output_type -> google.protobuf.Empty - 21, // 48: daemon.StartedService.SubscribeConnections:output_type -> daemon.Connections + 21, // 48: daemon.StartedService.SubscribeConnections:output_type -> daemon.ConnectionEvents 29, // 49: daemon.StartedService.CloseConnection:output_type -> google.protobuf.Empty 29, // 50: daemon.StartedService.CloseAllConnections:output_type -> google.protobuf.Empty 25, // 51: daemon.StartedService.GetDeprecatedWarnings:output_type -> daemon.DeprecatedWarnings @@ -2027,8 +2056,8 @@ func file_daemon_started_service_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_daemon_started_service_proto_rawDesc), len(file_daemon_started_service_proto_rawDesc)), - NumEnums: 4, - NumMessages: 25, + NumEnums: 3, + NumMessages: 26, NumExtensions: 0, NumServices: 1, }, diff --git a/daemon/started_service.proto b/daemon/started_service.proto index cd501cb5d..cc778f915 100644 --- a/daemon/started_service.proto +++ b/daemon/started_service.proto @@ -27,7 +27,7 @@ service StartedService { rpc GetSystemProxyStatus(google.protobuf.Empty) returns(SystemProxyStatus) {} rpc SetSystemProxyEnabled(SetSystemProxyEnabledRequest) returns(google.protobuf.Empty) {} - rpc SubscribeConnections(SubscribeConnectionsRequest) returns(stream Connections) {} + rpc SubscribeConnections(SubscribeConnectionsRequest) returns(stream ConnectionEvents) {} rpc CloseConnection(CloseConnectionRequest) returns(google.protobuf.Empty) {} rpc CloseAllConnections(google.protobuf.Empty) returns(google.protobuf.Empty) {} rpc GetDeprecatedWarnings(google.protobuf.Empty) returns(DeprecatedWarnings) {} @@ -143,24 +143,26 @@ message SetSystemProxyEnabledRequest { message SubscribeConnectionsRequest { int64 interval = 1; - ConnectionFilter filter = 2; - ConnectionSortBy sortBy = 3; } -enum ConnectionFilter { - ALL = 0; - ACTIVE = 1; - CLOSED = 2; +enum ConnectionEventType { + CONNECTION_EVENT_NEW = 0; + CONNECTION_EVENT_UPDATE = 1; + CONNECTION_EVENT_CLOSED = 2; } -enum ConnectionSortBy { - DATE = 0; - TRAFFIC = 1; - TOTAL_TRAFFIC = 2; +message ConnectionEvent { + ConnectionEventType type = 1; + string id = 2; + Connection connection = 3; + int64 uplinkDelta = 4; + int64 downlinkDelta = 5; + int64 closedAt = 6; } -message Connections { - repeated Connection connections = 1; +message ConnectionEvents { + repeated ConnectionEvent events = 1; + bool reset = 2; } message Connection { diff --git a/daemon/started_service_grpc.pb.go b/daemon/started_service_grpc.pb.go index 1fd09e405..438cca5c3 100644 --- a/daemon/started_service_grpc.pb.go +++ b/daemon/started_service_grpc.pb.go @@ -58,7 +58,7 @@ type StartedServiceClient interface { SetGroupExpand(ctx context.Context, in *SetGroupExpandRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) GetSystemProxyStatus(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*SystemProxyStatus, error) SetSystemProxyEnabled(ctx context.Context, in *SetSystemProxyEnabledRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) - SubscribeConnections(ctx context.Context, in *SubscribeConnectionsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Connections], error) + SubscribeConnections(ctx context.Context, in *SubscribeConnectionsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ConnectionEvents], error) CloseConnection(ctx context.Context, in *CloseConnectionRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) CloseAllConnections(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*emptypb.Empty, error) GetDeprecatedWarnings(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*DeprecatedWarnings, error) @@ -278,13 +278,13 @@ func (c *startedServiceClient) SetSystemProxyEnabled(ctx context.Context, in *Se return out, nil } -func (c *startedServiceClient) SubscribeConnections(ctx context.Context, in *SubscribeConnectionsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Connections], error) { +func (c *startedServiceClient) SubscribeConnections(ctx context.Context, in *SubscribeConnectionsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ConnectionEvents], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) stream, err := c.cc.NewStream(ctx, &StartedService_ServiceDesc.Streams[5], StartedService_SubscribeConnections_FullMethodName, cOpts...) if err != nil { return nil, err } - x := &grpc.GenericClientStream[SubscribeConnectionsRequest, Connections]{ClientStream: stream} + x := &grpc.GenericClientStream[SubscribeConnectionsRequest, ConnectionEvents]{ClientStream: stream} if err := x.ClientStream.SendMsg(in); err != nil { return nil, err } @@ -295,7 +295,7 @@ func (c *startedServiceClient) SubscribeConnections(ctx context.Context, in *Sub } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type StartedService_SubscribeConnectionsClient = grpc.ServerStreamingClient[Connections] +type StartedService_SubscribeConnectionsClient = grpc.ServerStreamingClient[ConnectionEvents] func (c *startedServiceClient) CloseConnection(ctx context.Context, in *CloseConnectionRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) @@ -357,7 +357,7 @@ type StartedServiceServer interface { SetGroupExpand(context.Context, *SetGroupExpandRequest) (*emptypb.Empty, error) GetSystemProxyStatus(context.Context, *emptypb.Empty) (*SystemProxyStatus, error) SetSystemProxyEnabled(context.Context, *SetSystemProxyEnabledRequest) (*emptypb.Empty, error) - SubscribeConnections(*SubscribeConnectionsRequest, grpc.ServerStreamingServer[Connections]) error + SubscribeConnections(*SubscribeConnectionsRequest, grpc.ServerStreamingServer[ConnectionEvents]) error CloseConnection(context.Context, *CloseConnectionRequest) (*emptypb.Empty, error) CloseAllConnections(context.Context, *emptypb.Empty) (*emptypb.Empty, error) GetDeprecatedWarnings(context.Context, *emptypb.Empty) (*DeprecatedWarnings, error) @@ -373,87 +373,87 @@ type StartedServiceServer interface { type UnimplementedStartedServiceServer struct{} func (UnimplementedStartedServiceServer) StopService(context.Context, *emptypb.Empty) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method StopService not implemented") + return nil, status.Error(codes.Unimplemented, "method StopService not implemented") } func (UnimplementedStartedServiceServer) ReloadService(context.Context, *emptypb.Empty) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method ReloadService not implemented") + return nil, status.Error(codes.Unimplemented, "method ReloadService not implemented") } func (UnimplementedStartedServiceServer) SubscribeServiceStatus(*emptypb.Empty, grpc.ServerStreamingServer[ServiceStatus]) error { - return status.Errorf(codes.Unimplemented, "method SubscribeServiceStatus not implemented") + return status.Error(codes.Unimplemented, "method SubscribeServiceStatus not implemented") } func (UnimplementedStartedServiceServer) SubscribeLog(*emptypb.Empty, grpc.ServerStreamingServer[Log]) error { - return status.Errorf(codes.Unimplemented, "method SubscribeLog not implemented") + return status.Error(codes.Unimplemented, "method SubscribeLog not implemented") } func (UnimplementedStartedServiceServer) GetDefaultLogLevel(context.Context, *emptypb.Empty) (*DefaultLogLevel, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetDefaultLogLevel not implemented") + return nil, status.Error(codes.Unimplemented, "method GetDefaultLogLevel not implemented") } func (UnimplementedStartedServiceServer) ClearLogs(context.Context, *emptypb.Empty) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method ClearLogs not implemented") + return nil, status.Error(codes.Unimplemented, "method ClearLogs not implemented") } func (UnimplementedStartedServiceServer) SubscribeStatus(*SubscribeStatusRequest, grpc.ServerStreamingServer[Status]) error { - return status.Errorf(codes.Unimplemented, "method SubscribeStatus not implemented") + return status.Error(codes.Unimplemented, "method SubscribeStatus not implemented") } func (UnimplementedStartedServiceServer) SubscribeGroups(*emptypb.Empty, grpc.ServerStreamingServer[Groups]) error { - return status.Errorf(codes.Unimplemented, "method SubscribeGroups not implemented") + return status.Error(codes.Unimplemented, "method SubscribeGroups not implemented") } func (UnimplementedStartedServiceServer) GetClashModeStatus(context.Context, *emptypb.Empty) (*ClashModeStatus, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetClashModeStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method GetClashModeStatus not implemented") } func (UnimplementedStartedServiceServer) SubscribeClashMode(*emptypb.Empty, grpc.ServerStreamingServer[ClashMode]) error { - return status.Errorf(codes.Unimplemented, "method SubscribeClashMode not implemented") + return status.Error(codes.Unimplemented, "method SubscribeClashMode not implemented") } func (UnimplementedStartedServiceServer) SetClashMode(context.Context, *ClashMode) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method SetClashMode not implemented") + return nil, status.Error(codes.Unimplemented, "method SetClashMode not implemented") } func (UnimplementedStartedServiceServer) URLTest(context.Context, *URLTestRequest) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method URLTest not implemented") + return nil, status.Error(codes.Unimplemented, "method URLTest not implemented") } func (UnimplementedStartedServiceServer) SelectOutbound(context.Context, *SelectOutboundRequest) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method SelectOutbound not implemented") + return nil, status.Error(codes.Unimplemented, "method SelectOutbound not implemented") } func (UnimplementedStartedServiceServer) SetGroupExpand(context.Context, *SetGroupExpandRequest) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method SetGroupExpand not implemented") + return nil, status.Error(codes.Unimplemented, "method SetGroupExpand not implemented") } func (UnimplementedStartedServiceServer) GetSystemProxyStatus(context.Context, *emptypb.Empty) (*SystemProxyStatus, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetSystemProxyStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method GetSystemProxyStatus not implemented") } func (UnimplementedStartedServiceServer) SetSystemProxyEnabled(context.Context, *SetSystemProxyEnabledRequest) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method SetSystemProxyEnabled not implemented") + return nil, status.Error(codes.Unimplemented, "method SetSystemProxyEnabled not implemented") } -func (UnimplementedStartedServiceServer) SubscribeConnections(*SubscribeConnectionsRequest, grpc.ServerStreamingServer[Connections]) error { - return status.Errorf(codes.Unimplemented, "method SubscribeConnections not implemented") +func (UnimplementedStartedServiceServer) SubscribeConnections(*SubscribeConnectionsRequest, grpc.ServerStreamingServer[ConnectionEvents]) error { + return status.Error(codes.Unimplemented, "method SubscribeConnections not implemented") } func (UnimplementedStartedServiceServer) CloseConnection(context.Context, *CloseConnectionRequest) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method CloseConnection not implemented") + return nil, status.Error(codes.Unimplemented, "method CloseConnection not implemented") } func (UnimplementedStartedServiceServer) CloseAllConnections(context.Context, *emptypb.Empty) (*emptypb.Empty, error) { - return nil, status.Errorf(codes.Unimplemented, "method CloseAllConnections not implemented") + return nil, status.Error(codes.Unimplemented, "method CloseAllConnections not implemented") } func (UnimplementedStartedServiceServer) GetDeprecatedWarnings(context.Context, *emptypb.Empty) (*DeprecatedWarnings, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetDeprecatedWarnings not implemented") + return nil, status.Error(codes.Unimplemented, "method GetDeprecatedWarnings not implemented") } func (UnimplementedStartedServiceServer) GetStartedAt(context.Context, *emptypb.Empty) (*StartedAt, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetStartedAt not implemented") + return nil, status.Error(codes.Unimplemented, "method GetStartedAt not implemented") } func (UnimplementedStartedServiceServer) mustEmbedUnimplementedStartedServiceServer() {} func (UnimplementedStartedServiceServer) testEmbeddedByValue() {} @@ -466,7 +466,7 @@ type UnsafeStartedServiceServer interface { } func RegisterStartedServiceServer(s grpc.ServiceRegistrar, srv StartedServiceServer) { - // If the following call pancis, it indicates UnimplementedStartedServiceServer was + // If the following call panics, it indicates UnimplementedStartedServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. @@ -734,11 +734,11 @@ func _StartedService_SubscribeConnections_Handler(srv interface{}, stream grpc.S if err := stream.RecvMsg(m); err != nil { return err } - return srv.(StartedServiceServer).SubscribeConnections(m, &grpc.GenericServerStream[SubscribeConnectionsRequest, Connections]{ServerStream: stream}) + return srv.(StartedServiceServer).SubscribeConnections(m, &grpc.GenericServerStream[SubscribeConnectionsRequest, ConnectionEvents]{ServerStream: stream}) } // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. -type StartedService_SubscribeConnectionsServer = grpc.ServerStreamingServer[Connections] +type StartedService_SubscribeConnectionsServer = grpc.ServerStreamingServer[ConnectionEvents] func _StartedService_CloseConnection_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(CloseConnectionRequest) diff --git a/experimental/clashapi/trafficontrol/manager.go b/experimental/clashapi/trafficontrol/manager.go index bb4822dff..45781c513 100644 --- a/experimental/clashapi/trafficontrol/manager.go +++ b/experimental/clashapi/trafficontrol/manager.go @@ -10,11 +10,29 @@ import ( C "github.com/sagernet/sing-box/constant" "github.com/sagernet/sing/common" "github.com/sagernet/sing/common/json" + "github.com/sagernet/sing/common/observable" "github.com/sagernet/sing/common/x/list" "github.com/gofrs/uuid/v5" ) +type ConnectionEventType int + +const ( + ConnectionEventNew ConnectionEventType = iota + ConnectionEventUpdate + ConnectionEventClosed +) + +type ConnectionEvent struct { + Type ConnectionEventType + ID uuid.UUID + Metadata TrackerMetadata + UplinkDelta int64 + DownlinkDelta int64 + ClosedAt time.Time +} + type Manager struct { uploadTotal atomic.Int64 downloadTotal atomic.Int64 @@ -22,16 +40,29 @@ type Manager struct { connections compatible.Map[uuid.UUID, Tracker] closedConnectionsAccess sync.Mutex closedConnections list.List[TrackerMetadata] - // process *process.Process - memory uint64 + memory uint64 + + eventSubscriber *observable.Subscriber[ConnectionEvent] } func NewManager() *Manager { return &Manager{} } +func (m *Manager) SetEventHook(subscriber *observable.Subscriber[ConnectionEvent]) { + m.eventSubscriber = subscriber +} + func (m *Manager) Join(c Tracker) { - m.connections.Store(c.Metadata().ID, c) + metadata := c.Metadata() + m.connections.Store(metadata.ID, c) + if m.eventSubscriber != nil { + m.eventSubscriber.Emit(ConnectionEvent{ + Type: ConnectionEventNew, + ID: metadata.ID, + Metadata: metadata, + }) + } } func (m *Manager) Leave(c Tracker) { @@ -40,11 +71,19 @@ func (m *Manager) Leave(c Tracker) { if loaded { metadata.ClosedAt = time.Now() m.closedConnectionsAccess.Lock() - defer m.closedConnectionsAccess.Unlock() if m.closedConnections.Len() >= 1000 { m.closedConnections.PopFront() } m.closedConnections.PushBack(metadata) + m.closedConnectionsAccess.Unlock() + if m.eventSubscriber != nil { + m.eventSubscriber.Emit(ConnectionEvent{ + Type: ConnectionEventClosed, + ID: metadata.ID, + Metadata: metadata, + ClosedAt: metadata.ClosedAt, + }) + } } } diff --git a/experimental/libbox/command_client.go b/experimental/libbox/command_client.go index 6f8b5accd..f0788d7ac 100644 --- a/experimental/libbox/command_client.go +++ b/experimental/libbox/command_client.go @@ -27,6 +27,7 @@ type CommandClient struct { ctx context.Context cancel context.CancelFunc clientMutex sync.RWMutex + standalone bool } type CommandClientOptions struct { @@ -48,7 +49,7 @@ type CommandClientHandler interface { WriteGroups(message OutboundGroupIterator) InitializeClashMode(modeList StringIterator, currentMode string) UpdateClashMode(newMode string) - WriteConnections(message *Connections) + WriteConnectionEvents(events *ConnectionEvents) } type LogEntry struct { @@ -73,7 +74,7 @@ func SetXPCDialer(dialer XPCDialer) { } func NewStandaloneCommandClient() *CommandClient { - return new(CommandClient) + return &CommandClient{standalone: true} } func NewCommandClient(handler CommandClientHandler, options *CommandClientOptions) *CommandClient { @@ -97,147 +98,135 @@ func streamClientAuthInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc return streamer(ctx, desc, cc, method, opts...) } -func (c *CommandClient) grpcDial() (*grpc.ClientConn, error) { - var target string - if sCommandServerListenPort == 0 { - target = "unix://" + filepath.Join(sBasePath, "command.sock") - } else { - target = net.JoinHostPort("127.0.0.1", strconv.Itoa(int(sCommandServerListenPort))) - } - var ( - conn *grpc.ClientConn - err error - ) - clientOptions := []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithUnaryInterceptor(unaryClientAuthInterceptor), - grpc.WithStreamInterceptor(streamClientAuthInterceptor), - } - for i := 0; i < 10; i++ { - conn, err = grpc.NewClient(target, clientOptions...) - if err == nil { - return conn, nil +const ( + commandClientDialAttempts = 10 + commandClientDialBaseDelay = 100 * time.Millisecond + commandClientDialStepDelay = 50 * time.Millisecond +) + +func commandClientDialDelay(attempt int) time.Duration { + return commandClientDialBaseDelay + time.Duration(attempt)*commandClientDialStepDelay +} + +func dialTarget() (string, func(context.Context, string) (net.Conn, error)) { + if sXPCDialer != nil { + return "passthrough:///xpc", func(ctx context.Context, _ string) (net.Conn, error) { + fileDescriptor, err := sXPCDialer.DialXPC() + if err != nil { + return nil, err + } + return networkConnectionFromFileDescriptor(fileDescriptor) } - time.Sleep(time.Duration(100+i*50) * time.Millisecond) } - return nil, err + if sCommandServerListenPort == 0 { + return "unix://" + filepath.Join(sBasePath, "command.sock"), nil + } + return net.JoinHostPort("127.0.0.1", strconv.Itoa(int(sCommandServerListenPort))), nil +} + +func networkConnectionFromFileDescriptor(fileDescriptor int32) (net.Conn, error) { + file := os.NewFile(uintptr(fileDescriptor), "xpc-command-socket") + if file == nil { + return nil, E.New("invalid file descriptor") + } + networkConnection, err := net.FileConn(file) + if err != nil { + file.Close() + return nil, E.Cause(err, "create connection from fd") + } + file.Close() + return networkConnection, nil +} + +func (c *CommandClient) dialWithRetry(target string, contextDialer func(context.Context, string) (net.Conn, error), retryDial bool) (*grpc.ClientConn, daemon.StartedServiceClient, error) { + var connection *grpc.ClientConn + var client daemon.StartedServiceClient + var lastError error + + for attempt := 0; attempt < commandClientDialAttempts; attempt++ { + if connection == nil { + options := []grpc.DialOption{ + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithUnaryInterceptor(unaryClientAuthInterceptor), + grpc.WithStreamInterceptor(streamClientAuthInterceptor), + } + if contextDialer != nil { + options = append(options, grpc.WithContextDialer(contextDialer)) + } + var err error + connection, err = grpc.NewClient(target, options...) + if err != nil { + lastError = err + if !retryDial { + return nil, nil, err + } + time.Sleep(commandClientDialDelay(attempt)) + continue + } + client = daemon.NewStartedServiceClient(connection) + } + waitDuration := commandClientDialDelay(attempt) + ctx, cancel := context.WithTimeout(context.Background(), waitDuration) + _, err := client.GetStartedAt(ctx, &emptypb.Empty{}, grpc.WaitForReady(true)) + cancel() + if err == nil { + return connection, client, nil + } + lastError = err + } + + if connection != nil { + connection.Close() + } + return nil, nil, lastError } func (c *CommandClient) Connect() error { c.clientMutex.Lock() common.Close(common.PtrOrNil(c.grpcConn)) - if sXPCDialer != nil { - fd, err := sXPCDialer.DialXPC() - if err != nil { - c.clientMutex.Unlock() - return err - } - file := os.NewFile(uintptr(fd), "xpc-command-socket") - if file == nil { - c.clientMutex.Unlock() - return E.New("invalid file descriptor") - } - netConn, err := net.FileConn(file) - if err != nil { - file.Close() - c.clientMutex.Unlock() - return E.Cause(err, "create connection from fd") - } - file.Close() - - clientOptions := []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { - return netConn, nil - }), - grpc.WithUnaryInterceptor(unaryClientAuthInterceptor), - grpc.WithStreamInterceptor(streamClientAuthInterceptor), - } - - grpcConn, err := grpc.NewClient("passthrough:///xpc", clientOptions...) - if err != nil { - netConn.Close() - c.clientMutex.Unlock() - return err - } - - c.grpcConn = grpcConn - c.grpcClient = daemon.NewStartedServiceClient(grpcConn) - c.ctx, c.cancel = context.WithCancel(context.Background()) - c.clientMutex.Unlock() - } else { - conn, err := c.grpcDial() - if err != nil { - c.clientMutex.Unlock() - return err - } - c.grpcConn = conn - c.grpcClient = daemon.NewStartedServiceClient(conn) - c.ctx, c.cancel = context.WithCancel(context.Background()) + target, contextDialer := dialTarget() + connection, client, err := c.dialWithRetry(target, contextDialer, true) + if err != nil { c.clientMutex.Unlock() + return err } + c.grpcConn = connection + c.grpcClient = client + c.ctx, c.cancel = context.WithCancel(context.Background()) + c.clientMutex.Unlock() c.handler.Connected() - for _, command := range c.options.commands { - switch command { - case CommandLog: - go c.handleLogStream() - case CommandStatus: - go c.handleStatusStream() - case CommandGroup: - go c.handleGroupStream() - case CommandClashMode: - go c.handleClashModeStream() - case CommandConnections: - go c.handleConnectionsStream() - default: - return E.New("unknown command: ", command) - } - } - return nil + return c.dispatchCommands() } func (c *CommandClient) ConnectWithFD(fd int32) error { c.clientMutex.Lock() common.Close(common.PtrOrNil(c.grpcConn)) - file := os.NewFile(uintptr(fd), "xpc-command-socket") - if file == nil { - c.clientMutex.Unlock() - return E.New("invalid file descriptor") - } - - netConn, err := net.FileConn(file) + networkConnection, err := networkConnectionFromFileDescriptor(fd) if err != nil { - file.Close() - c.clientMutex.Unlock() - return E.Cause(err, "create connection from fd") - } - file.Close() - - clientOptions := []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { - return netConn, nil - }), - grpc.WithUnaryInterceptor(unaryClientAuthInterceptor), - grpc.WithStreamInterceptor(streamClientAuthInterceptor), - } - - grpcConn, err := grpc.NewClient("passthrough:///xpc", clientOptions...) - if err != nil { - netConn.Close() c.clientMutex.Unlock() return err } - - c.grpcConn = grpcConn - c.grpcClient = daemon.NewStartedServiceClient(grpcConn) + connection, client, err := c.dialWithRetry("passthrough:///xpc", func(ctx context.Context, _ string) (net.Conn, error) { + return networkConnection, nil + }, false) + if err != nil { + networkConnection.Close() + c.clientMutex.Unlock() + return err + } + c.grpcConn = connection + c.grpcClient = client c.ctx, c.cancel = context.WithCancel(context.Background()) c.clientMutex.Unlock() c.handler.Connected() + return c.dispatchCommands() +} + +func (c *CommandClient) dispatchCommands() error { for _, command := range c.options.commands { switch command { case CommandLog: @@ -281,57 +270,41 @@ func (c *CommandClient) getClientForCall() (daemon.StartedServiceClient, error) return c.grpcClient, nil } - if sXPCDialer != nil { - fd, err := sXPCDialer.DialXPC() - if err != nil { - return nil, err - } - file := os.NewFile(uintptr(fd), "xpc-command-socket") - if file == nil { - return nil, E.New("invalid file descriptor") - } - netConn, err := net.FileConn(file) - if err != nil { - file.Close() - return nil, E.Cause(err, "create connection from fd") - } - file.Close() - - clientOptions := []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { - return netConn, nil - }), - grpc.WithUnaryInterceptor(unaryClientAuthInterceptor), - grpc.WithStreamInterceptor(streamClientAuthInterceptor), - } - - grpcConn, err := grpc.NewClient("passthrough:///xpc", clientOptions...) - if err != nil { - netConn.Close() - return nil, err - } - - c.grpcConn = grpcConn - c.grpcClient = daemon.NewStartedServiceClient(grpcConn) - if c.ctx == nil { - c.ctx, c.cancel = context.WithCancel(context.Background()) - } - return c.grpcClient, nil - } - - conn, err := c.grpcDial() + target, contextDialer := dialTarget() + connection, client, err := c.dialWithRetry(target, contextDialer, true) if err != nil { return nil, err } - c.grpcConn = conn - c.grpcClient = daemon.NewStartedServiceClient(conn) + c.grpcConn = connection + c.grpcClient = client if c.ctx == nil { c.ctx, c.cancel = context.WithCancel(context.Background()) } return c.grpcClient, nil } +func (c *CommandClient) closeConnection() { + c.clientMutex.Lock() + defer c.clientMutex.Unlock() + if c.grpcConn != nil { + c.grpcConn.Close() + c.grpcConn = nil + c.grpcClient = nil + } +} + +func callWithResult[T any](c *CommandClient, call func(client daemon.StartedServiceClient) (T, error)) (T, error) { + client, err := c.getClientForCall() + if err != nil { + var zero T + return zero, err + } + if c.standalone { + defer c.closeConnection() + } + return call(client) +} + func (c *CommandClient) getStreamContext() (daemon.StartedServiceClient, context.Context) { c.clientMutex.RLock() defer c.clientMutex.RUnlock() @@ -468,175 +441,134 @@ func (c *CommandClient) handleConnectionsStream() { return } - var connections Connections for { - conns, err := stream.Recv() + events, err := stream.Recv() if err != nil { c.handler.Disconnected(err.Error()) return } - connections.input = ConnectionsFromGRPC(conns) - c.handler.WriteConnections(&connections) + libboxEvents := ConnectionEventsFromGRPC(events) + c.handler.WriteConnectionEvents(libboxEvents) } } func (c *CommandClient) SelectOutbound(groupTag string, outboundTag string) error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.SelectOutbound(context.Background(), &daemon.SelectOutboundRequest{ - GroupTag: groupTag, - OutboundTag: outboundTag, + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.SelectOutbound(context.Background(), &daemon.SelectOutboundRequest{ + GroupTag: groupTag, + OutboundTag: outboundTag, + }) }) return err } func (c *CommandClient) URLTest(groupTag string) error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.URLTest(context.Background(), &daemon.URLTestRequest{ - OutboundTag: groupTag, + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.URLTest(context.Background(), &daemon.URLTestRequest{ + OutboundTag: groupTag, + }) }) return err } func (c *CommandClient) SetClashMode(newMode string) error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.SetClashMode(context.Background(), &daemon.ClashMode{ - Mode: newMode, + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.SetClashMode(context.Background(), &daemon.ClashMode{ + Mode: newMode, + }) }) return err } func (c *CommandClient) CloseConnection(connId string) error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.CloseConnection(context.Background(), &daemon.CloseConnectionRequest{ - Id: connId, + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.CloseConnection(context.Background(), &daemon.CloseConnectionRequest{ + Id: connId, + }) }) return err } func (c *CommandClient) CloseConnections() error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.CloseAllConnections(context.Background(), &emptypb.Empty{}) + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.CloseAllConnections(context.Background(), &emptypb.Empty{}) + }) return err } func (c *CommandClient) ServiceReload() error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.ReloadService(context.Background(), &emptypb.Empty{}) + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.ReloadService(context.Background(), &emptypb.Empty{}) + }) return err } func (c *CommandClient) ServiceClose() error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.StopService(context.Background(), &emptypb.Empty{}) + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.StopService(context.Background(), &emptypb.Empty{}) + }) return err } func (c *CommandClient) ClearLogs() error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.ClearLogs(context.Background(), &emptypb.Empty{}) + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.ClearLogs(context.Background(), &emptypb.Empty{}) + }) return err } func (c *CommandClient) GetSystemProxyStatus() (*SystemProxyStatus, error) { - client, err := c.getClientForCall() - if err != nil { - return nil, err - } - - status, err := client.GetSystemProxyStatus(context.Background(), &emptypb.Empty{}) - if err != nil { - return nil, err - } - return SystemProxyStatusFromGRPC(status), nil + return callWithResult(c, func(client daemon.StartedServiceClient) (*SystemProxyStatus, error) { + status, err := client.GetSystemProxyStatus(context.Background(), &emptypb.Empty{}) + if err != nil { + return nil, err + } + return SystemProxyStatusFromGRPC(status), nil + }) } func (c *CommandClient) SetSystemProxyEnabled(isEnabled bool) error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.SetSystemProxyEnabled(context.Background(), &daemon.SetSystemProxyEnabledRequest{ - Enabled: isEnabled, + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.SetSystemProxyEnabled(context.Background(), &daemon.SetSystemProxyEnabledRequest{ + Enabled: isEnabled, + }) }) return err } func (c *CommandClient) GetDeprecatedNotes() (DeprecatedNoteIterator, error) { - client, err := c.getClientForCall() - if err != nil { - return nil, err - } - - warnings, err := client.GetDeprecatedWarnings(context.Background(), &emptypb.Empty{}) - if err != nil { - return nil, err - } - - var notes []*DeprecatedNote - for _, warning := range warnings.Warnings { - notes = append(notes, &DeprecatedNote{ - Description: warning.Message, - MigrationLink: warning.MigrationLink, - }) - } - return newIterator(notes), nil + return callWithResult(c, func(client daemon.StartedServiceClient) (DeprecatedNoteIterator, error) { + warnings, err := client.GetDeprecatedWarnings(context.Background(), &emptypb.Empty{}) + if err != nil { + return nil, err + } + var notes []*DeprecatedNote + for _, warning := range warnings.Warnings { + notes = append(notes, &DeprecatedNote{ + Description: warning.Message, + MigrationLink: warning.MigrationLink, + }) + } + return newIterator(notes), nil + }) } func (c *CommandClient) GetStartedAt() (int64, error) { - client, err := c.getClientForCall() - if err != nil { - return 0, err - } - - startedAt, err := client.GetStartedAt(context.Background(), &emptypb.Empty{}) - if err != nil { - return 0, err - } - return startedAt.StartedAt, nil + return callWithResult(c, func(client daemon.StartedServiceClient) (int64, error) { + startedAt, err := client.GetStartedAt(context.Background(), &emptypb.Empty{}) + if err != nil { + return 0, err + } + return startedAt.StartedAt, nil + }) } func (c *CommandClient) SetGroupExpand(groupTag string, isExpand bool) error { - client, err := c.getClientForCall() - if err != nil { - return err - } - - _, err = client.SetGroupExpand(context.Background(), &daemon.SetGroupExpandRequest{ - GroupTag: groupTag, - IsExpand: isExpand, + _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { + return client.SetGroupExpand(context.Background(), &daemon.SetGroupExpandRequest{ + GroupTag: groupTag, + IsExpand: isExpand, + }) }) return err } diff --git a/experimental/libbox/command_types.go b/experimental/libbox/command_types.go index aa2fcdf23..1091b3f7d 100644 --- a/experimental/libbox/command_types.go +++ b/experimental/libbox/command_types.go @@ -3,6 +3,7 @@ package libbox import ( "slices" "strings" + "time" "github.com/sagernet/sing-box/daemon" M "github.com/sagernet/sing/common/metadata" @@ -61,12 +62,119 @@ const ( ConnectionStateClosed ) +const ( + ConnectionEventNew = iota + ConnectionEventUpdate + ConnectionEventClosed +) + +const ( + closedConnectionMaxAge = int64((5 * time.Minute) / time.Millisecond) +) + +type ConnectionEvent struct { + Type int32 + ID string + Connection *Connection + UplinkDelta int64 + DownlinkDelta int64 + ClosedAt int64 +} + +type ConnectionEvents struct { + Reset bool + events []*ConnectionEvent +} + +func (c *ConnectionEvents) Iterator() ConnectionEventIterator { + return newIterator(c.events) +} + +type ConnectionEventIterator interface { + Next() *ConnectionEvent + HasNext() bool +} + type Connections struct { - input []Connection - filtered []Connection + connectionMap map[string]*Connection + input []Connection + filtered []Connection + filterState int32 + filterApplied bool +} + +func NewConnections() *Connections { + return &Connections{ + connectionMap: make(map[string]*Connection), + } +} + +func (c *Connections) ApplyEvents(events *ConnectionEvents) { + if events == nil { + return + } + if events.Reset { + c.connectionMap = make(map[string]*Connection) + } + + for _, event := range events.events { + switch event.Type { + case ConnectionEventNew: + if event.Connection != nil { + conn := *event.Connection + c.connectionMap[event.ID] = &conn + } + case ConnectionEventUpdate: + if conn, ok := c.connectionMap[event.ID]; ok { + conn.Uplink = event.UplinkDelta + conn.Downlink = event.DownlinkDelta + conn.UplinkTotal += event.UplinkDelta + conn.DownlinkTotal += event.DownlinkDelta + } + case ConnectionEventClosed: + if event.Connection != nil { + conn := *event.Connection + conn.ClosedAt = event.ClosedAt + conn.Uplink = 0 + conn.Downlink = 0 + c.connectionMap[event.ID] = &conn + continue + } + if conn, ok := c.connectionMap[event.ID]; ok { + conn.ClosedAt = event.ClosedAt + conn.Uplink = 0 + conn.Downlink = 0 + } + } + } + + c.evictClosedConnections(time.Now().UnixMilli()) + c.input = c.input[:0] + for _, conn := range c.connectionMap { + c.input = append(c.input, *conn) + } + if c.filterApplied { + c.FilterState(c.filterState) + } else { + c.filtered = c.filtered[:0] + c.filtered = append(c.filtered, c.input...) + } +} + +func (c *Connections) evictClosedConnections(nowMilliseconds int64) { + for id, conn := range c.connectionMap { + if conn.ClosedAt == 0 { + continue + } + if nowMilliseconds-conn.ClosedAt > closedConnectionMaxAge { + delete(c.connectionMap, id) + } + } } func (c *Connections) FilterState(state int32) { + c.filterApplied = true + c.filterState = state c.filtered = c.filtered[:0] switch state { case ConnectionStateAll: @@ -264,15 +372,37 @@ func ConnectionFromGRPC(conn *daemon.Connection) Connection { } } -func ConnectionsFromGRPC(connections *daemon.Connections) []Connection { - if connections == nil || len(connections.Connections) == 0 { +func ConnectionEventFromGRPC(event *daemon.ConnectionEvent) *ConnectionEvent { + if event == nil { return nil } - var libboxConnections []Connection - for _, conn := range connections.Connections { - libboxConnections = append(libboxConnections, ConnectionFromGRPC(conn)) + libboxEvent := &ConnectionEvent{ + Type: int32(event.Type), + ID: event.Id, + UplinkDelta: event.UplinkDelta, + DownlinkDelta: event.DownlinkDelta, + ClosedAt: event.ClosedAt, } - return libboxConnections + if event.Connection != nil { + conn := ConnectionFromGRPC(event.Connection) + libboxEvent.Connection = &conn + } + return libboxEvent +} + +func ConnectionEventsFromGRPC(events *daemon.ConnectionEvents) *ConnectionEvents { + if events == nil { + return nil + } + libboxEvents := &ConnectionEvents{ + Reset: events.Reset_, + } + for _, event := range events.Events { + if libboxEvent := ConnectionEventFromGRPC(event); libboxEvent != nil { + libboxEvents.events = append(libboxEvents.events, libboxEvent) + } + } + return libboxEvents } func SystemProxyStatusFromGRPC(status *daemon.SystemProxyStatus) *SystemProxyStatus { diff --git a/experimental/v2rayapi/stats_grpc.pb.go b/experimental/v2rayapi/stats_grpc.pb.go index 3788f5205..0745899f0 100644 --- a/experimental/v2rayapi/stats_grpc.pb.go +++ b/experimental/v2rayapi/stats_grpc.pb.go @@ -84,15 +84,15 @@ type StatsServiceServer interface { type UnimplementedStatsServiceServer struct{} func (UnimplementedStatsServiceServer) GetStats(context.Context, *GetStatsRequest) (*GetStatsResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetStats not implemented") + return nil, status.Error(codes.Unimplemented, "method GetStats not implemented") } func (UnimplementedStatsServiceServer) QueryStats(context.Context, *QueryStatsRequest) (*QueryStatsResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method QueryStats not implemented") + return nil, status.Error(codes.Unimplemented, "method QueryStats not implemented") } func (UnimplementedStatsServiceServer) GetSysStats(context.Context, *SysStatsRequest) (*SysStatsResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetSysStats not implemented") + return nil, status.Error(codes.Unimplemented, "method GetSysStats not implemented") } func (UnimplementedStatsServiceServer) mustEmbedUnimplementedStatsServiceServer() {} func (UnimplementedStatsServiceServer) testEmbeddedByValue() {} @@ -105,7 +105,7 @@ type UnsafeStatsServiceServer interface { } func RegisterStatsServiceServer(s grpc.ServiceRegistrar, srv StatsServiceServer) { - // If the following call pancis, it indicates UnimplementedStatsServiceServer was + // If the following call panics, it indicates UnimplementedStatsServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. diff --git a/transport/v2raygrpc/stream_grpc.pb.go b/transport/v2raygrpc/stream_grpc.pb.go index d602ec452..21cc32796 100644 --- a/transport/v2raygrpc/stream_grpc.pb.go +++ b/transport/v2raygrpc/stream_grpc.pb.go @@ -61,7 +61,7 @@ type GunServiceServer interface { type UnimplementedGunServiceServer struct{} func (UnimplementedGunServiceServer) Tun(grpc.BidiStreamingServer[Hunk, Hunk]) error { - return status.Errorf(codes.Unimplemented, "method Tun not implemented") + return status.Error(codes.Unimplemented, "method Tun not implemented") } func (UnimplementedGunServiceServer) mustEmbedUnimplementedGunServiceServer() {} func (UnimplementedGunServiceServer) testEmbeddedByValue() {} @@ -74,7 +74,7 @@ type UnsafeGunServiceServer interface { } func RegisterGunServiceServer(s grpc.ServiceRegistrar, srv GunServiceServer) { - // If the following call pancis, it indicates UnimplementedGunServiceServer was + // If the following call panics, it indicates UnimplementedGunServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O.