diff --git a/api.go b/api.go index 056ae35..6e64d1e 100644 --- a/api.go +++ b/api.go @@ -2,19 +2,48 @@ package m7s import ( "context" + "errors" + "net" + "net/http" + "strings" + "time" + "github.com/shirou/gopsutil/v3/cpu" + "github.com/shirou/gopsutil/v3/disk" + "github.com/shirou/gopsutil/v3/mem" + . "github.com/shirou/gopsutil/v3/net" "google.golang.org/protobuf/types/known/emptypb" - "m7s.live/m7s/v5/pkg" + "google.golang.org/protobuf/types/known/timestamppb" "m7s.live/m7s/v5/pb" + "m7s.live/m7s/v5/pkg" + "m7s.live/m7s/v5/pkg/util" ) -func (s *Server) StreamSnap(ctx context.Context, req *pb.StreamSnapRequest) (res *pb.StreamSnapShot, err error) { - result, err := s.Call(req) - if err != nil { - return nil, err +var localIP string + +func (s *Server) SysInfo(context.Context, *emptypb.Empty) (res *pb.SysInfoResponse, err error) { + if localIP == "" { + if conn, err := net.Dial("udp", "114.114.114.114:80"); err == nil { + localIP, _, _ = strings.Cut(conn.LocalAddr().String(), ":") + } } - puber := result.(*Publisher) - return puber.SnapShot(), nil + res = &pb.SysInfoResponse{ + Version: Version, + LocalIP: localIP, + StartTime: timestamppb.New(s.StartTime), + } + return +} + +func (s *Server) StreamSnap(ctx context.Context, req *pb.StreamSnapRequest) (res *pb.StreamSnapShot, err error) { + s.Call(func() { + if pub, ok := s.Streams.Get(req.StreamPath); ok { + res = pub.SnapShot() + } else { + err = pkg.ErrNotFound + } + }) + return } func (s *Server) Restart(ctx context.Context, req *pb.RequestWithId) (res *emptypb.Empty, err error) { @@ -34,15 +63,82 @@ func (s *Server) Shutdown(ctx context.Context, req *pb.RequestWithId) (res *empt } func (s *Server) StopSubscribe(ctx context.Context, req *pb.StopSubscribeRequest) (res *pb.StopSubscribeResponse, err error) { - _, err = s.Call(req) + s.Call(func() { + if subscriber, ok := s.Subscribers.Get(int(req.Id)); ok { + subscriber.Stop(errors.New("stop by api")) + } else { + err = pkg.ErrNotFound + } + }) return &pb.StopSubscribeResponse{ Success: err == nil, }, err } +func (s *Server) StreamList(_ context.Context, req *pb.StreamListRequest) (res *pb.StreamListResponse, err error) { + s.Call(func() { + var streams []*pb.StreamSummay + for _, publisher := range s.Streams.Items { + streams = append(streams, &pb.StreamSummay{ + Path: publisher.StreamPath, + }) + } + res = &pb.StreamListResponse{List: streams, Total: int32(s.Streams.Length), PageNum: req.PageNum, PageSize: req.PageSize} + }) + return +} -func (s *Server) StreamList(ctx context.Context, req *pb.StreamListRequest) (res *pb.StreamListResponse, err error) { - var result any - result, err = s.Call(req) - return result.(*pb.StreamListResponse), err -} \ No newline at end of file +func (s *Server) API_Summary_SSE(rw http.ResponseWriter, r *http.Request) { + util.ReturnFetchValue(func() *pb.SummaryResponse { + ret, _ := s.Summary(r.Context(), nil) + return ret + }, rw, r) +} + +func (s *Server) Summary(context.Context, *emptypb.Empty) (res *pb.SummaryResponse, err error) { + s.Call(func() { + dur := time.Since(s.lastSummaryTime) + if dur < time.Second { + res = s.lastSummary + return + } + v, _ := mem.VirtualMemory() + d, _ := disk.Usage("/") + nv, _ := IOCounters(true) + res = &pb.SummaryResponse{ + Memory: &pb.Usage{ + Total: v.Total >> 20, + Free: v.Available >> 20, + Used: v.Used >> 20, + Usage: float32(v.UsedPercent), + }, + HardDisk: &pb.Usage{ + Total: d.Total >> 30, + Free: d.Free >> 30, + Used: d.Used >> 30, + Usage: float32(d.UsedPercent), + }, + } + if cc, _ := cpu.Percent(time.Second, false); len(cc) > 0 { + res.CpuUsage = float32(cc[0]) + } + netWorks := []*pb.NetWorkInfo{} + for i, n := range nv { + info := &pb.NetWorkInfo{ + Name: n.Name, + Receive: n.BytesRecv, + Sent: n.BytesSent, + } + if s.lastSummary != nil && len(s.lastSummary.NetWork) > i { + info.ReceiveSpeed = (n.BytesRecv - s.lastSummary.NetWork[i].Receive) / uint64(dur.Seconds()) + info.SentSpeed = (n.BytesSent - s.lastSummary.NetWork[i].Sent) / uint64(dur.Seconds()) + } + netWorks = append(netWorks, info) + } + res.StreamCount = int32(s.Streams.Length) + res.NetWork = netWorks + s.lastSummary = res + s.lastSummaryTime = time.Now() + }) + return +} diff --git a/example/default/config.yaml b/example/default/config.yaml index d92c7a6..bd46e60 100644 --- a/example/default/config.yaml +++ b/example/default/config.yaml @@ -1,9 +1,11 @@ global: - loglevel: trace + loglevel: debug tcp: listenaddr: :50051 +console: + secret: de2c0bb9fd47684adc07a426e139239b logrotate: - level: trace + level: info webrtc: publish: pubaudio: false diff --git a/example/default/fatal/latest.log b/example/default/fatal/latest.log new file mode 100644 index 0000000..e69de29 diff --git a/example/default/main.go b/example/default/main.go index 560c51a..fb81b4a 100644 --- a/example/default/main.go +++ b/example/default/main.go @@ -9,6 +9,7 @@ import ( _ "m7s.live/m7s/v5/plugin/webrtc" _ "m7s.live/m7s/v5/plugin/rtmp" _ "m7s.live/m7s/v5/plugin/logrotate" + _ "m7s.live/m7s/v5/plugin/console" ) func main() { diff --git a/pb/global.pb.go b/pb/global.pb.go index 0d9db9b..ad3ff3a 100644 --- a/pb/global.pb.go +++ b/pb/global.pb.go @@ -11,6 +11,7 @@ import ( protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" emptypb "google.golang.org/protobuf/types/known/emptypb" + timestamppb "google.golang.org/protobuf/types/known/timestamppb" reflect "reflect" sync "sync" ) @@ -22,16 +23,20 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -type StreamInfo struct { +type NetWorkInfo struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + Receive uint64 `protobuf:"varint,2,opt,name=receive,proto3" json:"receive,omitempty"` + Sent uint64 `protobuf:"varint,3,opt,name=sent,proto3" json:"sent,omitempty"` + ReceiveSpeed uint64 `protobuf:"varint,4,opt,name=receiveSpeed,proto3" json:"receiveSpeed,omitempty"` + SentSpeed uint64 `protobuf:"varint,5,opt,name=sentSpeed,proto3" json:"sentSpeed,omitempty"` } -func (x *StreamInfo) Reset() { - *x = StreamInfo{} +func (x *NetWorkInfo) Reset() { + *x = NetWorkInfo{} if protoimpl.UnsafeEnabled { mi := &file_global_proto_msgTypes[0] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -39,13 +44,13 @@ func (x *StreamInfo) Reset() { } } -func (x *StreamInfo) String() string { +func (x *NetWorkInfo) String() string { return protoimpl.X.MessageStringOf(x) } -func (*StreamInfo) ProtoMessage() {} +func (*NetWorkInfo) ProtoMessage() {} -func (x *StreamInfo) ProtoReflect() protoreflect.Message { +func (x *NetWorkInfo) ProtoReflect() protoreflect.Message { mi := &file_global_proto_msgTypes[0] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -57,28 +62,375 @@ func (x *StreamInfo) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use StreamInfo.ProtoReflect.Descriptor instead. -func (*StreamInfo) Descriptor() ([]byte, []int) { +// Deprecated: Use NetWorkInfo.ProtoReflect.Descriptor instead. +func (*NetWorkInfo) Descriptor() ([]byte, []int) { return file_global_proto_rawDescGZIP(), []int{0} } -func (x *StreamInfo) GetPath() string { +func (x *NetWorkInfo) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *NetWorkInfo) GetReceive() uint64 { + if x != nil { + return x.Receive + } + return 0 +} + +func (x *NetWorkInfo) GetSent() uint64 { + if x != nil { + return x.Sent + } + return 0 +} + +func (x *NetWorkInfo) GetReceiveSpeed() uint64 { + if x != nil { + return x.ReceiveSpeed + } + return 0 +} + +func (x *NetWorkInfo) GetSentSpeed() uint64 { + if x != nil { + return x.SentSpeed + } + return 0 +} + +type Usage struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Total uint64 `protobuf:"varint,1,opt,name=total,proto3" json:"total,omitempty"` + Free uint64 `protobuf:"varint,2,opt,name=free,proto3" json:"free,omitempty"` + Used uint64 `protobuf:"varint,3,opt,name=used,proto3" json:"used,omitempty"` + Usage float32 `protobuf:"fixed32,4,opt,name=usage,proto3" json:"usage,omitempty"` +} + +func (x *Usage) Reset() { + *x = Usage{} + if protoimpl.UnsafeEnabled { + mi := &file_global_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Usage) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Usage) ProtoMessage() {} + +func (x *Usage) ProtoReflect() protoreflect.Message { + mi := &file_global_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Usage.ProtoReflect.Descriptor instead. +func (*Usage) Descriptor() ([]byte, []int) { + return file_global_proto_rawDescGZIP(), []int{1} +} + +func (x *Usage) GetTotal() uint64 { + if x != nil { + return x.Total + } + return 0 +} + +func (x *Usage) GetFree() uint64 { + if x != nil { + return x.Free + } + return 0 +} + +func (x *Usage) GetUsed() uint64 { + if x != nil { + return x.Used + } + return 0 +} + +func (x *Usage) GetUsage() float32 { + if x != nil { + return x.Usage + } + return 0 +} + +type SummaryResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Address string `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"` + Memory *Usage `protobuf:"bytes,2,opt,name=memory,proto3" json:"memory,omitempty"` + CpuUsage float32 `protobuf:"fixed32,3,opt,name=cpuUsage,proto3" json:"cpuUsage,omitempty"` + HardDisk *Usage `protobuf:"bytes,4,opt,name=hardDisk,proto3" json:"hardDisk,omitempty"` + NetWork []*NetWorkInfo `protobuf:"bytes,5,rep,name=netWork,proto3" json:"netWork,omitempty"` + StreamCount int32 `protobuf:"varint,6,opt,name=streamCount,proto3" json:"streamCount,omitempty"` +} + +func (x *SummaryResponse) Reset() { + *x = SummaryResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_global_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *SummaryResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SummaryResponse) ProtoMessage() {} + +func (x *SummaryResponse) ProtoReflect() protoreflect.Message { + mi := &file_global_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SummaryResponse.ProtoReflect.Descriptor instead. +func (*SummaryResponse) Descriptor() ([]byte, []int) { + return file_global_proto_rawDescGZIP(), []int{2} +} + +func (x *SummaryResponse) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +func (x *SummaryResponse) GetMemory() *Usage { + if x != nil { + return x.Memory + } + return nil +} + +func (x *SummaryResponse) GetCpuUsage() float32 { + if x != nil { + return x.CpuUsage + } + return 0 +} + +func (x *SummaryResponse) GetHardDisk() *Usage { + if x != nil { + return x.HardDisk + } + return nil +} + +func (x *SummaryResponse) GetNetWork() []*NetWorkInfo { + if x != nil { + return x.NetWork + } + return nil +} + +func (x *SummaryResponse) GetStreamCount() int32 { + if x != nil { + return x.StreamCount + } + return 0 +} + +type StreamSummay struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"` + State int32 `protobuf:"varint,2,opt,name=state,proto3" json:"state,omitempty"` + Subscribers int32 `protobuf:"varint,3,opt,name=subscribers,proto3" json:"subscribers,omitempty"` + Tracks []string `protobuf:"bytes,4,rep,name=tracks,proto3" json:"tracks,omitempty"` + StartTime *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=startTime,proto3" json:"startTime,omitempty"` + Type string `protobuf:"bytes,6,opt,name=type,proto3" json:"type,omitempty"` + Bps int32 `protobuf:"varint,7,opt,name=bps,proto3" json:"bps,omitempty"` +} + +func (x *StreamSummay) Reset() { + *x = StreamSummay{} + if protoimpl.UnsafeEnabled { + mi := &file_global_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *StreamSummay) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamSummay) ProtoMessage() {} + +func (x *StreamSummay) ProtoReflect() protoreflect.Message { + mi := &file_global_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamSummay.ProtoReflect.Descriptor instead. +func (*StreamSummay) Descriptor() ([]byte, []int) { + return file_global_proto_rawDescGZIP(), []int{3} +} + +func (x *StreamSummay) GetPath() string { if x != nil { return x.Path } return "" } +func (x *StreamSummay) GetState() int32 { + if x != nil { + return x.State + } + return 0 +} + +func (x *StreamSummay) GetSubscribers() int32 { + if x != nil { + return x.Subscribers + } + return 0 +} + +func (x *StreamSummay) GetTracks() []string { + if x != nil { + return x.Tracks + } + return nil +} + +func (x *StreamSummay) GetStartTime() *timestamppb.Timestamp { + if x != nil { + return x.StartTime + } + return nil +} + +func (x *StreamSummay) GetType() string { + if x != nil { + return x.Type + } + return "" +} + +func (x *StreamSummay) GetBps() int32 { + if x != nil { + return x.Bps + } + return 0 +} + +type SysInfoResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + StartTime *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=startTime,proto3" json:"startTime,omitempty"` + LocalIP string `protobuf:"bytes,2,opt,name=localIP,proto3" json:"localIP,omitempty"` + Version string `protobuf:"bytes,3,opt,name=version,proto3" json:"version,omitempty"` +} + +func (x *SysInfoResponse) Reset() { + *x = SysInfoResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_global_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *SysInfoResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SysInfoResponse) ProtoMessage() {} + +func (x *SysInfoResponse) ProtoReflect() protoreflect.Message { + mi := &file_global_proto_msgTypes[4] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SysInfoResponse.ProtoReflect.Descriptor instead. +func (*SysInfoResponse) Descriptor() ([]byte, []int) { + return file_global_proto_rawDescGZIP(), []int{4} +} + +func (x *SysInfoResponse) GetStartTime() *timestamppb.Timestamp { + if x != nil { + return x.StartTime + } + return nil +} + +func (x *SysInfoResponse) GetLocalIP() string { + if x != nil { + return x.LocalIP + } + return "" +} + +func (x *SysInfoResponse) GetVersion() string { + if x != nil { + return x.Version + } + return "" +} + type StreamListRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields + + PageNum int32 `protobuf:"varint,1,opt,name=pageNum,proto3" json:"pageNum,omitempty"` + PageSize int32 `protobuf:"varint,2,opt,name=pageSize,proto3" json:"pageSize,omitempty"` } func (x *StreamListRequest) Reset() { *x = StreamListRequest{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[1] + mi := &file_global_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -91,7 +443,7 @@ func (x *StreamListRequest) String() string { func (*StreamListRequest) ProtoMessage() {} func (x *StreamListRequest) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[1] + mi := &file_global_proto_msgTypes[5] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -104,7 +456,21 @@ func (x *StreamListRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamListRequest.ProtoReflect.Descriptor instead. func (*StreamListRequest) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{1} + return file_global_proto_rawDescGZIP(), []int{5} +} + +func (x *StreamListRequest) GetPageNum() int32 { + if x != nil { + return x.PageNum + } + return 0 +} + +func (x *StreamListRequest) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 } type StreamListResponse struct { @@ -112,13 +478,16 @@ type StreamListResponse struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - List []*StreamInfo `protobuf:"bytes,1,rep,name=list,proto3" json:"list,omitempty"` + Total int32 `protobuf:"varint,1,opt,name=total,proto3" json:"total,omitempty"` + PageNum int32 `protobuf:"varint,2,opt,name=pageNum,proto3" json:"pageNum,omitempty"` + PageSize int32 `protobuf:"varint,3,opt,name=pageSize,proto3" json:"pageSize,omitempty"` + List []*StreamSummay `protobuf:"bytes,4,rep,name=list,proto3" json:"list,omitempty"` } func (x *StreamListResponse) Reset() { *x = StreamListResponse{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[2] + mi := &file_global_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -131,7 +500,7 @@ func (x *StreamListResponse) String() string { func (*StreamListResponse) ProtoMessage() {} func (x *StreamListResponse) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[2] + mi := &file_global_proto_msgTypes[6] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -144,10 +513,31 @@ func (x *StreamListResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamListResponse.ProtoReflect.Descriptor instead. func (*StreamListResponse) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{2} + return file_global_proto_rawDescGZIP(), []int{6} } -func (x *StreamListResponse) GetList() []*StreamInfo { +func (x *StreamListResponse) GetTotal() int32 { + if x != nil { + return x.Total + } + return 0 +} + +func (x *StreamListResponse) GetPageNum() int32 { + if x != nil { + return x.PageNum + } + return 0 +} + +func (x *StreamListResponse) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 +} + +func (x *StreamListResponse) GetList() []*StreamSummay { if x != nil { return x.List } @@ -165,7 +555,7 @@ type StreamSnapRequest struct { func (x *StreamSnapRequest) Reset() { *x = StreamSnapRequest{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[3] + mi := &file_global_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -178,7 +568,7 @@ func (x *StreamSnapRequest) String() string { func (*StreamSnapRequest) ProtoMessage() {} func (x *StreamSnapRequest) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[3] + mi := &file_global_proto_msgTypes[7] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -191,7 +581,7 @@ func (x *StreamSnapRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamSnapRequest.ProtoReflect.Descriptor instead. func (*StreamSnapRequest) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{3} + return file_global_proto_rawDescGZIP(), []int{7} } func (x *StreamSnapRequest) GetStreamPath() string { @@ -214,7 +604,7 @@ type Wrap struct { func (x *Wrap) Reset() { *x = Wrap{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[4] + mi := &file_global_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -227,7 +617,7 @@ func (x *Wrap) String() string { func (*Wrap) ProtoMessage() {} func (x *Wrap) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[4] + mi := &file_global_proto_msgTypes[8] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -240,7 +630,7 @@ func (x *Wrap) ProtoReflect() protoreflect.Message { // Deprecated: Use Wrap.ProtoReflect.Descriptor instead. func (*Wrap) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{4} + return file_global_proto_rawDescGZIP(), []int{8} } func (x *Wrap) GetTimestamp() uint32 { @@ -279,7 +669,7 @@ type TrackSnapShot struct { func (x *TrackSnapShot) Reset() { *x = TrackSnapShot{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[5] + mi := &file_global_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -292,7 +682,7 @@ func (x *TrackSnapShot) String() string { func (*TrackSnapShot) ProtoMessage() {} func (x *TrackSnapShot) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[5] + mi := &file_global_proto_msgTypes[9] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -305,7 +695,7 @@ func (x *TrackSnapShot) ProtoReflect() protoreflect.Message { // Deprecated: Use TrackSnapShot.ProtoReflect.Descriptor instead. func (*TrackSnapShot) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{5} + return file_global_proto_rawDescGZIP(), []int{9} } func (x *TrackSnapShot) GetSequence() uint32 { @@ -355,7 +745,7 @@ type StreamSnapShot struct { func (x *StreamSnapShot) Reset() { *x = StreamSnapShot{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[6] + mi := &file_global_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -368,7 +758,7 @@ func (x *StreamSnapShot) String() string { func (*StreamSnapShot) ProtoMessage() {} func (x *StreamSnapShot) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[6] + mi := &file_global_proto_msgTypes[10] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -381,7 +771,7 @@ func (x *StreamSnapShot) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamSnapShot.ProtoReflect.Descriptor instead. func (*StreamSnapShot) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{6} + return file_global_proto_rawDescGZIP(), []int{10} } func (x *StreamSnapShot) GetVideoTrack() []*TrackSnapShot { @@ -409,7 +799,7 @@ type StopSubscribeRequest struct { func (x *StopSubscribeRequest) Reset() { *x = StopSubscribeRequest{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[7] + mi := &file_global_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -422,7 +812,7 @@ func (x *StopSubscribeRequest) String() string { func (*StopSubscribeRequest) ProtoMessage() {} func (x *StopSubscribeRequest) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[7] + mi := &file_global_proto_msgTypes[11] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -435,7 +825,7 @@ func (x *StopSubscribeRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use StopSubscribeRequest.ProtoReflect.Descriptor instead. func (*StopSubscribeRequest) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{7} + return file_global_proto_rawDescGZIP(), []int{11} } func (x *StopSubscribeRequest) GetId() uint32 { @@ -456,7 +846,7 @@ type StopSubscribeResponse struct { func (x *StopSubscribeResponse) Reset() { *x = StopSubscribeResponse{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[8] + mi := &file_global_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -469,7 +859,7 @@ func (x *StopSubscribeResponse) String() string { func (*StopSubscribeResponse) ProtoMessage() {} func (x *StopSubscribeResponse) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[8] + mi := &file_global_proto_msgTypes[12] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -482,7 +872,7 @@ func (x *StopSubscribeResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use StopSubscribeResponse.ProtoReflect.Descriptor instead. func (*StopSubscribeResponse) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{8} + return file_global_proto_rawDescGZIP(), []int{12} } func (x *StopSubscribeResponse) GetSuccess() bool { @@ -503,7 +893,7 @@ type RequestWithId struct { func (x *RequestWithId) Reset() { *x = RequestWithId{} if protoimpl.UnsafeEnabled { - mi := &file_global_proto_msgTypes[9] + mi := &file_global_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -516,7 +906,7 @@ func (x *RequestWithId) String() string { func (*RequestWithId) ProtoMessage() {} func (x *RequestWithId) ProtoReflect() protoreflect.Message { - mi := &file_global_proto_msgTypes[9] + mi := &file_global_proto_msgTypes[13] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -529,7 +919,7 @@ func (x *RequestWithId) ProtoReflect() protoreflect.Message { // Deprecated: Use RequestWithId.ProtoReflect.Descriptor instead. func (*RequestWithId) Descriptor() ([]byte, []int) { - return file_global_proto_rawDescGZIP(), []int{9} + return file_global_proto_rawDescGZIP(), []int{13} } func (x *RequestWithId) GetId() uint32 { @@ -546,80 +936,149 @@ var file_global_proto_rawDesc = []byte{ 0x6d, 0x37, 0x73, 0x1a, 0x1c, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x61, 0x6e, 0x6e, 0x6f, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1b, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, - 0x75, 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x20, - 0x0a, 0x0a, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x12, 0x0a, 0x04, - 0x70, 0x61, 0x74, 0x68, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68, - 0x22, 0x13, 0x0a, 0x11, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x39, 0x0a, 0x12, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4c, - 0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a, 0x04, 0x6c, - 0x69, 0x73, 0x74, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x6d, 0x37, 0x73, 0x2e, - 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x04, 0x6c, 0x69, 0x73, 0x74, - 0x22, 0x33, 0x0a, 0x11, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x6e, 0x61, 0x70, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x50, - 0x61, 0x74, 0x68, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x50, 0x61, 0x74, 0x68, 0x22, 0x4c, 0x0a, 0x04, 0x57, 0x72, 0x61, 0x70, 0x12, 0x1c, 0x0a, - 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, - 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x73, - 0x69, 0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x73, 0x69, 0x7a, 0x65, 0x12, - 0x12, 0x0a, 0x04, 0x64, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x64, - 0x61, 0x74, 0x61, 0x22, 0xa0, 0x01, 0x0a, 0x0d, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x6e, 0x61, - 0x70, 0x53, 0x68, 0x6f, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, - 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, - 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0d, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, - 0x1c, 0x0a, 0x09, 0x77, 0x72, 0x69, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, - 0x28, 0x04, 0x52, 0x09, 0x77, 0x72, 0x69, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x18, 0x0a, - 0x07, 0x63, 0x61, 0x6e, 0x52, 0x65, 0x61, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, - 0x63, 0x61, 0x6e, 0x52, 0x65, 0x61, 0x64, 0x12, 0x1d, 0x0a, 0x04, 0x77, 0x72, 0x61, 0x70, 0x18, - 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x57, 0x72, 0x61, 0x70, - 0x52, 0x04, 0x77, 0x72, 0x61, 0x70, 0x22, 0x78, 0x0a, 0x0e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, - 0x53, 0x6e, 0x61, 0x70, 0x53, 0x68, 0x6f, 0x74, 0x12, 0x32, 0x0a, 0x0a, 0x76, 0x69, 0x64, 0x65, - 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x6d, - 0x37, 0x73, 0x2e, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x6e, 0x61, 0x70, 0x53, 0x68, 0x6f, 0x74, - 0x52, 0x0a, 0x76, 0x69, 0x64, 0x65, 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x12, 0x32, 0x0a, 0x0a, - 0x61, 0x75, 0x64, 0x69, 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, - 0x32, 0x12, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x6e, 0x61, 0x70, - 0x53, 0x68, 0x6f, 0x74, 0x52, 0x0a, 0x61, 0x75, 0x64, 0x69, 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, - 0x22, 0x26, 0x0a, 0x14, 0x53, 0x74, 0x6f, 0x70, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, - 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, - 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x22, 0x31, 0x0a, 0x15, 0x53, 0x74, 0x6f, 0x70, - 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, - 0x65, 0x12, 0x18, 0x0a, 0x07, 0x73, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, - 0x28, 0x08, 0x52, 0x07, 0x73, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x22, 0x1f, 0x0a, 0x0d, 0x52, - 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x57, 0x69, 0x74, 0x68, 0x49, 0x64, 0x12, 0x0e, 0x0a, 0x02, - 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x32, 0xd9, 0x03, 0x0a, - 0x06, 0x47, 0x6c, 0x6f, 0x62, 0x61, 0x6c, 0x12, 0x52, 0x0a, 0x08, 0x53, 0x68, 0x75, 0x74, 0x64, - 0x6f, 0x77, 0x6e, 0x12, 0x12, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x75, 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f, + 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, + 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, + 0x91, 0x01, 0x0a, 0x0b, 0x4e, 0x65, 0x74, 0x57, 0x6f, 0x72, 0x6b, 0x49, 0x6e, 0x66, 0x6f, 0x12, + 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, + 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x72, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x72, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x12, 0x12, 0x0a, + 0x04, 0x73, 0x65, 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x52, 0x04, 0x73, 0x65, 0x6e, + 0x74, 0x12, 0x22, 0x0a, 0x0c, 0x72, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x53, 0x70, 0x65, 0x65, + 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x72, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, + 0x53, 0x70, 0x65, 0x65, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x73, 0x65, 0x6e, 0x74, 0x53, 0x70, 0x65, + 0x65, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x73, 0x65, 0x6e, 0x74, 0x53, 0x70, + 0x65, 0x65, 0x64, 0x22, 0x5b, 0x0a, 0x05, 0x55, 0x73, 0x61, 0x67, 0x65, 0x12, 0x14, 0x0a, 0x05, + 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x05, 0x74, 0x6f, 0x74, + 0x61, 0x6c, 0x12, 0x12, 0x0a, 0x04, 0x66, 0x72, 0x65, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, + 0x52, 0x04, 0x66, 0x72, 0x65, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x73, 0x65, 0x64, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x04, 0x52, 0x04, 0x75, 0x73, 0x65, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x75, 0x73, + 0x61, 0x67, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x02, 0x52, 0x05, 0x75, 0x73, 0x61, 0x67, 0x65, + 0x22, 0xe1, 0x01, 0x0a, 0x0f, 0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x22, + 0x0a, 0x06, 0x6d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, + 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x55, 0x73, 0x61, 0x67, 0x65, 0x52, 0x06, 0x6d, 0x65, 0x6d, 0x6f, + 0x72, 0x79, 0x12, 0x1a, 0x0a, 0x08, 0x63, 0x70, 0x75, 0x55, 0x73, 0x61, 0x67, 0x65, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x02, 0x52, 0x08, 0x63, 0x70, 0x75, 0x55, 0x73, 0x61, 0x67, 0x65, 0x12, 0x26, + 0x0a, 0x08, 0x68, 0x61, 0x72, 0x64, 0x44, 0x69, 0x73, 0x6b, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, + 0x32, 0x0a, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x55, 0x73, 0x61, 0x67, 0x65, 0x52, 0x08, 0x68, 0x61, + 0x72, 0x64, 0x44, 0x69, 0x73, 0x6b, 0x12, 0x2a, 0x0a, 0x07, 0x6e, 0x65, 0x74, 0x57, 0x6f, 0x72, + 0x6b, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x4e, 0x65, + 0x74, 0x57, 0x6f, 0x72, 0x6b, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x07, 0x6e, 0x65, 0x74, 0x57, 0x6f, + 0x72, 0x6b, 0x12, 0x20, 0x0a, 0x0b, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x43, 0x6f, 0x75, 0x6e, + 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x43, + 0x6f, 0x75, 0x6e, 0x74, 0x22, 0xd2, 0x01, 0x0a, 0x0c, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, + 0x75, 0x6d, 0x6d, 0x61, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x12, 0x14, 0x0a, 0x05, 0x73, 0x74, 0x61, + 0x74, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x05, 0x73, 0x74, 0x61, 0x74, 0x65, 0x12, + 0x20, 0x0a, 0x0b, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x73, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, + 0x73, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, + 0x09, 0x52, 0x06, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x73, 0x12, 0x38, 0x0a, 0x09, 0x73, 0x74, 0x61, + 0x72, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, + 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, + 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x73, 0x74, 0x61, 0x72, 0x74, 0x54, + 0x69, 0x6d, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x62, 0x70, 0x73, 0x18, 0x07, + 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x62, 0x70, 0x73, 0x22, 0x7f, 0x0a, 0x0f, 0x53, 0x79, 0x73, + 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x38, 0x0a, 0x09, + 0x73, 0x74, 0x61, 0x72, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, + 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, + 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x73, 0x74, 0x61, + 0x72, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x6c, 0x6f, 0x63, 0x61, 0x6c, 0x49, + 0x50, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6c, 0x6f, 0x63, 0x61, 0x6c, 0x49, 0x50, + 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x49, 0x0a, 0x11, 0x53, 0x74, + 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, + 0x18, 0x0a, 0x07, 0x70, 0x61, 0x67, 0x65, 0x4e, 0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, + 0x52, 0x07, 0x70, 0x61, 0x67, 0x65, 0x4e, 0x75, 0x6d, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x67, + 0x65, 0x53, 0x69, 0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x70, 0x61, 0x67, + 0x65, 0x53, 0x69, 0x7a, 0x65, 0x22, 0x87, 0x01, 0x0a, 0x12, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, + 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x05, 0x74, 0x6f, 0x74, + 0x61, 0x6c, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x67, 0x65, 0x4e, 0x75, 0x6d, 0x18, 0x02, 0x20, + 0x01, 0x28, 0x05, 0x52, 0x07, 0x70, 0x61, 0x67, 0x65, 0x4e, 0x75, 0x6d, 0x12, 0x1a, 0x0a, 0x08, + 0x70, 0x61, 0x67, 0x65, 0x53, 0x69, 0x7a, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, + 0x70, 0x61, 0x67, 0x65, 0x53, 0x69, 0x7a, 0x65, 0x12, 0x25, 0x0a, 0x04, 0x6c, 0x69, 0x73, 0x74, + 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x11, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, 0x72, + 0x65, 0x61, 0x6d, 0x53, 0x75, 0x6d, 0x6d, 0x61, 0x79, 0x52, 0x04, 0x6c, 0x69, 0x73, 0x74, 0x22, + 0x33, 0x0a, 0x11, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x6e, 0x61, 0x70, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x50, 0x61, + 0x74, 0x68, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x50, 0x61, 0x74, 0x68, 0x22, 0x4c, 0x0a, 0x04, 0x57, 0x72, 0x61, 0x70, 0x12, 0x1c, 0x0a, 0x09, + 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, + 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x73, 0x69, + 0x7a, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x73, 0x69, 0x7a, 0x65, 0x12, 0x12, + 0x0a, 0x04, 0x64, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x64, 0x61, + 0x74, 0x61, 0x22, 0xa0, 0x01, 0x0a, 0x0d, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x6e, 0x61, 0x70, + 0x53, 0x68, 0x6f, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, 0x65, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, 0x65, + 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, + 0x01, 0x28, 0x0d, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x1c, + 0x0a, 0x09, 0x77, 0x72, 0x69, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, + 0x04, 0x52, 0x09, 0x77, 0x72, 0x69, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, + 0x63, 0x61, 0x6e, 0x52, 0x65, 0x61, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, 0x63, + 0x61, 0x6e, 0x52, 0x65, 0x61, 0x64, 0x12, 0x1d, 0x0a, 0x04, 0x77, 0x72, 0x61, 0x70, 0x18, 0x05, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x57, 0x72, 0x61, 0x70, 0x52, + 0x04, 0x77, 0x72, 0x61, 0x70, 0x22, 0x78, 0x0a, 0x0e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, + 0x6e, 0x61, 0x70, 0x53, 0x68, 0x6f, 0x74, 0x12, 0x32, 0x0a, 0x0a, 0x76, 0x69, 0x64, 0x65, 0x6f, + 0x54, 0x72, 0x61, 0x63, 0x6b, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x6d, 0x37, + 0x73, 0x2e, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x6e, 0x61, 0x70, 0x53, 0x68, 0x6f, 0x74, 0x52, + 0x0a, 0x76, 0x69, 0x64, 0x65, 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x12, 0x32, 0x0a, 0x0a, 0x61, + 0x75, 0x64, 0x69, 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, + 0x12, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x53, 0x6e, 0x61, 0x70, 0x53, + 0x68, 0x6f, 0x74, 0x52, 0x0a, 0x61, 0x75, 0x64, 0x69, 0x6f, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x22, + 0x26, 0x0a, 0x14, 0x53, 0x74, 0x6f, 0x70, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x22, 0x31, 0x0a, 0x15, 0x53, 0x74, 0x6f, 0x70, 0x53, + 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x12, 0x18, 0x0a, 0x07, 0x73, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x08, 0x52, 0x07, 0x73, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x22, 0x1f, 0x0a, 0x0d, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x57, 0x69, 0x74, 0x68, 0x49, 0x64, 0x12, 0x0e, 0x0a, 0x02, 0x69, + 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x32, 0xf7, 0x04, 0x0a, 0x06, + 0x47, 0x6c, 0x6f, 0x62, 0x61, 0x6c, 0x12, 0x4d, 0x0a, 0x07, 0x53, 0x79, 0x73, 0x49, 0x6e, 0x66, + 0x6f, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x14, 0x2e, 0x6d, 0x37, 0x73, 0x2e, + 0x53, 0x79, 0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, + 0x14, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x0e, 0x12, 0x0c, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x79, + 0x73, 0x69, 0x6e, 0x66, 0x6f, 0x12, 0x4d, 0x0a, 0x07, 0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, + 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, + 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x14, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, + 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x14, + 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x0e, 0x12, 0x0c, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x75, 0x6d, + 0x6d, 0x61, 0x72, 0x79, 0x12, 0x52, 0x0a, 0x08, 0x53, 0x68, 0x75, 0x74, 0x64, 0x6f, 0x77, 0x6e, + 0x12, 0x12, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x57, 0x69, + 0x74, 0x68, 0x49, 0x64, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, + 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x1a, 0x82, 0xd3, + 0xe4, 0x93, 0x02, 0x14, 0x22, 0x12, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x68, 0x75, 0x74, 0x64, + 0x6f, 0x77, 0x6e, 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x12, 0x50, 0x0a, 0x07, 0x52, 0x65, 0x73, 0x74, + 0x61, 0x72, 0x74, 0x12, 0x12, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x57, 0x69, 0x74, 0x68, 0x49, 0x64, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, - 0x1a, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x14, 0x22, 0x12, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x68, - 0x75, 0x74, 0x64, 0x6f, 0x77, 0x6e, 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x12, 0x50, 0x0a, 0x07, 0x52, - 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x12, 0x12, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x52, 0x65, 0x71, - 0x75, 0x65, 0x73, 0x74, 0x57, 0x69, 0x74, 0x68, 0x49, 0x64, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, - 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, - 0x74, 0x79, 0x22, 0x19, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x13, 0x22, 0x11, 0x2f, 0x61, 0x70, 0x69, - 0x2f, 0x72, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x12, 0x57, 0x0a, - 0x0a, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x16, 0x2e, 0x6d, 0x37, - 0x73, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, - 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, - 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x18, 0x82, 0xd3, - 0xe4, 0x93, 0x02, 0x12, 0x12, 0x10, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x2f, 0x6c, 0x69, 0x73, 0x74, 0x12, 0x63, 0x0a, 0x0a, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, - 0x53, 0x6e, 0x61, 0x70, 0x12, 0x16, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x53, 0x6e, 0x61, 0x70, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x6d, - 0x37, 0x73, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x6e, 0x61, 0x70, 0x53, 0x68, 0x6f, - 0x74, 0x22, 0x28, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x22, 0x12, 0x20, 0x2f, 0x61, 0x70, 0x69, 0x2f, - 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f, 0x73, 0x6e, 0x61, 0x70, 0x2f, 0x7b, 0x73, 0x74, 0x72, - 0x65, 0x61, 0x6d, 0x50, 0x61, 0x74, 0x68, 0x3d, 0x2a, 0x2a, 0x7d, 0x12, 0x6b, 0x0a, 0x0d, 0x53, - 0x74, 0x6f, 0x70, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x19, 0x2e, 0x6d, - 0x37, 0x73, 0x2e, 0x53, 0x74, 0x6f, 0x70, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, - 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, - 0x6f, 0x70, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, - 0x6e, 0x73, 0x65, 0x22, 0x23, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x1d, 0x22, 0x18, 0x2f, 0x61, 0x70, - 0x69, 0x2f, 0x73, 0x74, 0x6f, 0x70, 0x2f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, - 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x3a, 0x01, 0x2a, 0x42, 0x14, 0x5a, 0x12, 0x6d, 0x37, 0x73, 0x2e, - 0x6c, 0x69, 0x76, 0x65, 0x2f, 0x6d, 0x37, 0x73, 0x2f, 0x76, 0x35, 0x2f, 0x70, 0x62, 0x62, 0x06, - 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x19, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x13, 0x22, 0x11, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x72, 0x65, + 0x73, 0x74, 0x61, 0x72, 0x74, 0x2f, 0x7b, 0x69, 0x64, 0x7d, 0x12, 0x57, 0x0a, 0x0a, 0x53, 0x74, + 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x16, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, + 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x1a, 0x17, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4c, 0x69, 0x73, + 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x18, 0x82, 0xd3, 0xe4, 0x93, 0x02, + 0x12, 0x12, 0x10, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2f, 0x6c, + 0x69, 0x73, 0x74, 0x12, 0x63, 0x0a, 0x0a, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x6e, 0x61, + 0x70, 0x12, 0x16, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x6e, + 0x61, 0x70, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x6d, 0x37, 0x73, 0x2e, + 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x6e, 0x61, 0x70, 0x53, 0x68, 0x6f, 0x74, 0x22, 0x28, + 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x22, 0x12, 0x20, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, 0x74, 0x72, + 0x65, 0x61, 0x6d, 0x2f, 0x73, 0x6e, 0x61, 0x70, 0x2f, 0x7b, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x50, 0x61, 0x74, 0x68, 0x3d, 0x2a, 0x2a, 0x7d, 0x12, 0x6b, 0x0a, 0x0d, 0x53, 0x74, 0x6f, 0x70, + 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x19, 0x2e, 0x6d, 0x37, 0x73, 0x2e, + 0x53, 0x74, 0x6f, 0x70, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x6d, 0x37, 0x73, 0x2e, 0x53, 0x74, 0x6f, 0x70, 0x53, + 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x22, 0x23, 0x82, 0xd3, 0xe4, 0x93, 0x02, 0x1d, 0x22, 0x18, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x73, + 0x74, 0x6f, 0x70, 0x2f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x2f, 0x7b, 0x69, + 0x64, 0x7d, 0x3a, 0x01, 0x2a, 0x42, 0x14, 0x5a, 0x12, 0x6d, 0x37, 0x73, 0x2e, 0x6c, 0x69, 0x76, + 0x65, 0x2f, 0x6d, 0x37, 0x73, 0x2f, 0x76, 0x35, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, + 0x74, 0x6f, 0x33, } var ( @@ -634,40 +1093,54 @@ func file_global_proto_rawDescGZIP() []byte { return file_global_proto_rawDescData } -var file_global_proto_msgTypes = make([]protoimpl.MessageInfo, 10) +var file_global_proto_msgTypes = make([]protoimpl.MessageInfo, 14) var file_global_proto_goTypes = []interface{}{ - (*StreamInfo)(nil), // 0: m7s.StreamInfo - (*StreamListRequest)(nil), // 1: m7s.StreamListRequest - (*StreamListResponse)(nil), // 2: m7s.StreamListResponse - (*StreamSnapRequest)(nil), // 3: m7s.StreamSnapRequest - (*Wrap)(nil), // 4: m7s.Wrap - (*TrackSnapShot)(nil), // 5: m7s.TrackSnapShot - (*StreamSnapShot)(nil), // 6: m7s.StreamSnapShot - (*StopSubscribeRequest)(nil), // 7: m7s.StopSubscribeRequest - (*StopSubscribeResponse)(nil), // 8: m7s.StopSubscribeResponse - (*RequestWithId)(nil), // 9: m7s.RequestWithId - (*emptypb.Empty)(nil), // 10: google.protobuf.Empty + (*NetWorkInfo)(nil), // 0: m7s.NetWorkInfo + (*Usage)(nil), // 1: m7s.Usage + (*SummaryResponse)(nil), // 2: m7s.SummaryResponse + (*StreamSummay)(nil), // 3: m7s.StreamSummay + (*SysInfoResponse)(nil), // 4: m7s.SysInfoResponse + (*StreamListRequest)(nil), // 5: m7s.StreamListRequest + (*StreamListResponse)(nil), // 6: m7s.StreamListResponse + (*StreamSnapRequest)(nil), // 7: m7s.StreamSnapRequest + (*Wrap)(nil), // 8: m7s.Wrap + (*TrackSnapShot)(nil), // 9: m7s.TrackSnapShot + (*StreamSnapShot)(nil), // 10: m7s.StreamSnapShot + (*StopSubscribeRequest)(nil), // 11: m7s.StopSubscribeRequest + (*StopSubscribeResponse)(nil), // 12: m7s.StopSubscribeResponse + (*RequestWithId)(nil), // 13: m7s.RequestWithId + (*timestamppb.Timestamp)(nil), // 14: google.protobuf.Timestamp + (*emptypb.Empty)(nil), // 15: google.protobuf.Empty } var file_global_proto_depIdxs = []int32{ - 0, // 0: m7s.StreamListResponse.list:type_name -> m7s.StreamInfo - 4, // 1: m7s.TrackSnapShot.wrap:type_name -> m7s.Wrap - 5, // 2: m7s.StreamSnapShot.videoTrack:type_name -> m7s.TrackSnapShot - 5, // 3: m7s.StreamSnapShot.audioTrack:type_name -> m7s.TrackSnapShot - 9, // 4: m7s.Global.Shutdown:input_type -> m7s.RequestWithId - 9, // 5: m7s.Global.Restart:input_type -> m7s.RequestWithId - 1, // 6: m7s.Global.StreamList:input_type -> m7s.StreamListRequest - 3, // 7: m7s.Global.StreamSnap:input_type -> m7s.StreamSnapRequest - 7, // 8: m7s.Global.StopSubscribe:input_type -> m7s.StopSubscribeRequest - 10, // 9: m7s.Global.Shutdown:output_type -> google.protobuf.Empty - 10, // 10: m7s.Global.Restart:output_type -> google.protobuf.Empty - 2, // 11: m7s.Global.StreamList:output_type -> m7s.StreamListResponse - 6, // 12: m7s.Global.StreamSnap:output_type -> m7s.StreamSnapShot - 8, // 13: m7s.Global.StopSubscribe:output_type -> m7s.StopSubscribeResponse - 9, // [9:14] is the sub-list for method output_type - 4, // [4:9] is the sub-list for method input_type - 4, // [4:4] is the sub-list for extension type_name - 4, // [4:4] is the sub-list for extension extendee - 0, // [0:4] is the sub-list for field type_name + 1, // 0: m7s.SummaryResponse.memory:type_name -> m7s.Usage + 1, // 1: m7s.SummaryResponse.hardDisk:type_name -> m7s.Usage + 0, // 2: m7s.SummaryResponse.netWork:type_name -> m7s.NetWorkInfo + 14, // 3: m7s.StreamSummay.startTime:type_name -> google.protobuf.Timestamp + 14, // 4: m7s.SysInfoResponse.startTime:type_name -> google.protobuf.Timestamp + 3, // 5: m7s.StreamListResponse.list:type_name -> m7s.StreamSummay + 8, // 6: m7s.TrackSnapShot.wrap:type_name -> m7s.Wrap + 9, // 7: m7s.StreamSnapShot.videoTrack:type_name -> m7s.TrackSnapShot + 9, // 8: m7s.StreamSnapShot.audioTrack:type_name -> m7s.TrackSnapShot + 15, // 9: m7s.Global.SysInfo:input_type -> google.protobuf.Empty + 15, // 10: m7s.Global.Summary:input_type -> google.protobuf.Empty + 13, // 11: m7s.Global.Shutdown:input_type -> m7s.RequestWithId + 13, // 12: m7s.Global.Restart:input_type -> m7s.RequestWithId + 5, // 13: m7s.Global.StreamList:input_type -> m7s.StreamListRequest + 7, // 14: m7s.Global.StreamSnap:input_type -> m7s.StreamSnapRequest + 11, // 15: m7s.Global.StopSubscribe:input_type -> m7s.StopSubscribeRequest + 4, // 16: m7s.Global.SysInfo:output_type -> m7s.SysInfoResponse + 2, // 17: m7s.Global.Summary:output_type -> m7s.SummaryResponse + 15, // 18: m7s.Global.Shutdown:output_type -> google.protobuf.Empty + 15, // 19: m7s.Global.Restart:output_type -> google.protobuf.Empty + 6, // 20: m7s.Global.StreamList:output_type -> m7s.StreamListResponse + 10, // 21: m7s.Global.StreamSnap:output_type -> m7s.StreamSnapShot + 12, // 22: m7s.Global.StopSubscribe:output_type -> m7s.StopSubscribeResponse + 16, // [16:23] is the sub-list for method output_type + 9, // [9:16] is the sub-list for method input_type + 9, // [9:9] is the sub-list for extension type_name + 9, // [9:9] is the sub-list for extension extendee + 0, // [0:9] is the sub-list for field type_name } func init() { file_global_proto_init() } @@ -677,7 +1150,7 @@ func file_global_proto_init() { } if !protoimpl.UnsafeEnabled { file_global_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StreamInfo); i { + switch v := v.(*NetWorkInfo); i { case 0: return &v.state case 1: @@ -689,7 +1162,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StreamListRequest); i { + switch v := v.(*Usage); i { case 0: return &v.state case 1: @@ -701,7 +1174,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StreamListResponse); i { + switch v := v.(*SummaryResponse); i { case 0: return &v.state case 1: @@ -713,7 +1186,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StreamSnapRequest); i { + switch v := v.(*StreamSummay); i { case 0: return &v.state case 1: @@ -725,7 +1198,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*Wrap); i { + switch v := v.(*SysInfoResponse); i { case 0: return &v.state case 1: @@ -737,7 +1210,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*TrackSnapShot); i { + switch v := v.(*StreamListRequest); i { case 0: return &v.state case 1: @@ -749,7 +1222,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StreamSnapShot); i { + switch v := v.(*StreamListResponse); i { case 0: return &v.state case 1: @@ -761,7 +1234,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StopSubscribeRequest); i { + switch v := v.(*StreamSnapRequest); i { case 0: return &v.state case 1: @@ -773,7 +1246,7 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StopSubscribeResponse); i { + switch v := v.(*Wrap); i { case 0: return &v.state case 1: @@ -785,6 +1258,54 @@ func file_global_proto_init() { } } file_global_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*TrackSnapShot); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_global_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*StreamSnapShot); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_global_proto_msgTypes[11].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*StopSubscribeRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_global_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*StopSubscribeResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_global_proto_msgTypes[13].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*RequestWithId); i { case 0: return &v.state @@ -803,7 +1324,7 @@ func file_global_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_global_proto_rawDesc, NumEnums: 0, - NumMessages: 10, + NumMessages: 14, NumExtensions: 0, NumServices: 1, }, diff --git a/pb/global.pb.gw.go b/pb/global.pb.gw.go index 5bbc23b..87950e1 100644 --- a/pb/global.pb.gw.go +++ b/pb/global.pb.gw.go @@ -21,6 +21,7 @@ import ( "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/emptypb" ) // Suppress "imported and not used" errors @@ -31,6 +32,42 @@ var _ = runtime.String var _ = utilities.NewDoubleArray var _ = metadata.Join +func request_Global_SysInfo_0(ctx context.Context, marshaler runtime.Marshaler, client GlobalClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) { + var protoReq emptypb.Empty + var metadata runtime.ServerMetadata + + msg, err := client.SysInfo(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD)) + return msg, metadata, err + +} + +func local_request_Global_SysInfo_0(ctx context.Context, marshaler runtime.Marshaler, server GlobalServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) { + var protoReq emptypb.Empty + var metadata runtime.ServerMetadata + + msg, err := server.SysInfo(ctx, &protoReq) + return msg, metadata, err + +} + +func request_Global_Summary_0(ctx context.Context, marshaler runtime.Marshaler, client GlobalClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) { + var protoReq emptypb.Empty + var metadata runtime.ServerMetadata + + msg, err := client.Summary(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD)) + return msg, metadata, err + +} + +func local_request_Global_Summary_0(ctx context.Context, marshaler runtime.Marshaler, server GlobalServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) { + var protoReq emptypb.Empty + var metadata runtime.ServerMetadata + + msg, err := server.Summary(ctx, &protoReq) + return msg, metadata, err + +} + func request_Global_Shutdown_0(ctx context.Context, marshaler runtime.Marshaler, client GlobalClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) { var protoReq RequestWithId var metadata runtime.ServerMetadata @@ -135,10 +172,21 @@ func local_request_Global_Restart_0(ctx context.Context, marshaler runtime.Marsh } +var ( + filter_Global_StreamList_0 = &utilities.DoubleArray{Encoding: map[string]int{}, Base: []int(nil), Check: []int(nil)} +) + func request_Global_StreamList_0(ctx context.Context, marshaler runtime.Marshaler, client GlobalClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) { var protoReq StreamListRequest var metadata runtime.ServerMetadata + if err := req.ParseForm(); err != nil { + return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err) + } + if err := runtime.PopulateQueryParameters(&protoReq, req.Form, filter_Global_StreamList_0); err != nil { + return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err) + } + msg, err := client.StreamList(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD)) return msg, metadata, err @@ -148,6 +196,13 @@ func local_request_Global_StreamList_0(ctx context.Context, marshaler runtime.Ma var protoReq StreamListRequest var metadata runtime.ServerMetadata + if err := req.ParseForm(); err != nil { + return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err) + } + if err := runtime.PopulateQueryParameters(&protoReq, req.Form, filter_Global_StreamList_0); err != nil { + return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err) + } + msg, err := server.StreamList(ctx, &protoReq) return msg, metadata, err @@ -271,6 +326,56 @@ func local_request_Global_StopSubscribe_0(ctx context.Context, marshaler runtime // Note that using this registration option will cause many gRPC library features to stop working. Consider using RegisterGlobalHandlerFromEndpoint instead. func RegisterGlobalHandlerServer(ctx context.Context, mux *runtime.ServeMux, server GlobalServer) error { + mux.Handle("GET", pattern_Global_SysInfo_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) { + ctx, cancel := context.WithCancel(req.Context()) + defer cancel() + var stream runtime.ServerTransportStream + ctx = grpc.NewContextWithServerTransportStream(ctx, &stream) + inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req) + var err error + var annotatedContext context.Context + annotatedContext, err = runtime.AnnotateIncomingContext(ctx, mux, req, "/m7s.Global/SysInfo", runtime.WithHTTPPathPattern("/api/sysinfo")) + if err != nil { + runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err) + return + } + resp, md, err := local_request_Global_SysInfo_0(annotatedContext, inboundMarshaler, server, req, pathParams) + md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer()) + annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md) + if err != nil { + runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err) + return + } + + forward_Global_SysInfo_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...) + + }) + + mux.Handle("GET", pattern_Global_Summary_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) { + ctx, cancel := context.WithCancel(req.Context()) + defer cancel() + var stream runtime.ServerTransportStream + ctx = grpc.NewContextWithServerTransportStream(ctx, &stream) + inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req) + var err error + var annotatedContext context.Context + annotatedContext, err = runtime.AnnotateIncomingContext(ctx, mux, req, "/m7s.Global/Summary", runtime.WithHTTPPathPattern("/api/summary")) + if err != nil { + runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err) + return + } + resp, md, err := local_request_Global_Summary_0(annotatedContext, inboundMarshaler, server, req, pathParams) + md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer()) + annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md) + if err != nil { + runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err) + return + } + + forward_Global_Summary_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...) + + }) + mux.Handle("POST", pattern_Global_Shutdown_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) { ctx, cancel := context.WithCancel(req.Context()) defer cancel() @@ -437,6 +542,50 @@ func RegisterGlobalHandler(ctx context.Context, mux *runtime.ServeMux, conn *grp // "GlobalClient" to call the correct interceptors. func RegisterGlobalHandlerClient(ctx context.Context, mux *runtime.ServeMux, client GlobalClient) error { + mux.Handle("GET", pattern_Global_SysInfo_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) { + ctx, cancel := context.WithCancel(req.Context()) + defer cancel() + inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req) + var err error + var annotatedContext context.Context + annotatedContext, err = runtime.AnnotateContext(ctx, mux, req, "/m7s.Global/SysInfo", runtime.WithHTTPPathPattern("/api/sysinfo")) + if err != nil { + runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err) + return + } + resp, md, err := request_Global_SysInfo_0(annotatedContext, inboundMarshaler, client, req, pathParams) + annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md) + if err != nil { + runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err) + return + } + + forward_Global_SysInfo_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...) + + }) + + mux.Handle("GET", pattern_Global_Summary_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) { + ctx, cancel := context.WithCancel(req.Context()) + defer cancel() + inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req) + var err error + var annotatedContext context.Context + annotatedContext, err = runtime.AnnotateContext(ctx, mux, req, "/m7s.Global/Summary", runtime.WithHTTPPathPattern("/api/summary")) + if err != nil { + runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err) + return + } + resp, md, err := request_Global_Summary_0(annotatedContext, inboundMarshaler, client, req, pathParams) + annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md) + if err != nil { + runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err) + return + } + + forward_Global_Summary_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...) + + }) + mux.Handle("POST", pattern_Global_Shutdown_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) { ctx, cancel := context.WithCancel(req.Context()) defer cancel() @@ -551,6 +700,10 @@ func RegisterGlobalHandlerClient(ctx context.Context, mux *runtime.ServeMux, cli } var ( + pattern_Global_SysInfo_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1}, []string{"api", "sysinfo"}, "")) + + pattern_Global_Summary_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1}, []string{"api", "summary"}, "")) + pattern_Global_Shutdown_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2}, []string{"api", "shutdown", "id"}, "")) pattern_Global_Restart_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2}, []string{"api", "restart", "id"}, "")) @@ -563,6 +716,10 @@ var ( ) var ( + forward_Global_SysInfo_0 = runtime.ForwardResponseMessage + + forward_Global_Summary_0 = runtime.ForwardResponseMessage + forward_Global_Shutdown_0 = runtime.ForwardResponseMessage forward_Global_Restart_0 = runtime.ForwardResponseMessage diff --git a/pb/global.proto b/pb/global.proto index 00eb7ec..22d4434 100644 --- a/pb/global.proto +++ b/pb/global.proto @@ -1,10 +1,21 @@ syntax = "proto3"; import "google/api/annotations.proto"; import "google/protobuf/empty.proto"; +import "google/protobuf/timestamp.proto"; package m7s; option go_package="m7s.live/m7s/v5/pb"; service Global { + rpc SysInfo (google.protobuf.Empty) returns (SysInfoResponse) { + option (google.api.http) = { + get: "/api/sysinfo" + }; + } + rpc Summary (google.protobuf.Empty) returns (SummaryResponse) { + option (google.api.http) = { + get: "/api/summary" + }; + } rpc Shutdown (RequestWithId) returns (google.protobuf.Empty) { option (google.api.http) = { post: "/api/shutdown/{id}" @@ -32,14 +43,57 @@ service Global { }; } } -message StreamInfo { - string path = 1; -} -message StreamListRequest { +message NetWorkInfo { + string name = 1; + uint64 receive = 2; + uint64 sent = 3; + uint64 receiveSpeed = 4; + uint64 sentSpeed = 5; } + +message Usage { + uint64 total = 1; + uint64 free = 2; + uint64 used = 3; + float usage = 4; +} + +message SummaryResponse { + string address = 1; + Usage memory = 2; + float cpuUsage = 3; + Usage hardDisk = 4; + repeated NetWorkInfo netWork = 5; + int32 streamCount = 6; +} + +message StreamSummay { + string path = 1; + int32 state = 2; + int32 subscribers = 3; + repeated string tracks = 4; + google.protobuf.Timestamp startTime = 5; + string type = 6; + int32 bps = 7; +} + +message SysInfoResponse { + google.protobuf.Timestamp startTime = 1; + string localIP = 2; + string version = 3; +} + +message StreamListRequest { + int32 pageNum = 1; + int32 pageSize = 2; +} + message StreamListResponse { - repeated StreamInfo list = 1; + int32 total = 1; + int32 pageNum = 2; + int32 pageSize = 3; + repeated StreamSummay list = 4; } message StreamSnapRequest { diff --git a/pb/global_grpc.pb.go b/pb/global_grpc.pb.go index 1fb6c79..8d18ee3 100644 --- a/pb/global_grpc.pb.go +++ b/pb/global_grpc.pb.go @@ -23,6 +23,8 @@ const _ = grpc.SupportPackageIsVersion7 // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type GlobalClient interface { + SysInfo(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*SysInfoResponse, error) + Summary(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*SummaryResponse, error) Shutdown(ctx context.Context, in *RequestWithId, opts ...grpc.CallOption) (*emptypb.Empty, error) Restart(ctx context.Context, in *RequestWithId, opts ...grpc.CallOption) (*emptypb.Empty, error) StreamList(ctx context.Context, in *StreamListRequest, opts ...grpc.CallOption) (*StreamListResponse, error) @@ -38,6 +40,24 @@ func NewGlobalClient(cc grpc.ClientConnInterface) GlobalClient { return &globalClient{cc} } +func (c *globalClient) SysInfo(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*SysInfoResponse, error) { + out := new(SysInfoResponse) + err := c.cc.Invoke(ctx, "/m7s.Global/SysInfo", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *globalClient) Summary(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*SummaryResponse, error) { + out := new(SummaryResponse) + err := c.cc.Invoke(ctx, "/m7s.Global/Summary", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *globalClient) Shutdown(ctx context.Context, in *RequestWithId, opts ...grpc.CallOption) (*emptypb.Empty, error) { out := new(emptypb.Empty) err := c.cc.Invoke(ctx, "/m7s.Global/Shutdown", in, out, opts...) @@ -87,6 +107,8 @@ func (c *globalClient) StopSubscribe(ctx context.Context, in *StopSubscribeReque // All implementations must embed UnimplementedGlobalServer // for forward compatibility type GlobalServer interface { + SysInfo(context.Context, *emptypb.Empty) (*SysInfoResponse, error) + Summary(context.Context, *emptypb.Empty) (*SummaryResponse, error) Shutdown(context.Context, *RequestWithId) (*emptypb.Empty, error) Restart(context.Context, *RequestWithId) (*emptypb.Empty, error) StreamList(context.Context, *StreamListRequest) (*StreamListResponse, error) @@ -99,6 +121,12 @@ type GlobalServer interface { type UnimplementedGlobalServer struct { } +func (UnimplementedGlobalServer) SysInfo(context.Context, *emptypb.Empty) (*SysInfoResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method SysInfo not implemented") +} +func (UnimplementedGlobalServer) Summary(context.Context, *emptypb.Empty) (*SummaryResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method Summary not implemented") +} func (UnimplementedGlobalServer) Shutdown(context.Context, *RequestWithId) (*emptypb.Empty, error) { return nil, status.Errorf(codes.Unimplemented, "method Shutdown not implemented") } @@ -127,6 +155,42 @@ func RegisterGlobalServer(s grpc.ServiceRegistrar, srv GlobalServer) { s.RegisterService(&Global_ServiceDesc, srv) } +func _Global_SysInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(emptypb.Empty) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(GlobalServer).SysInfo(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/m7s.Global/SysInfo", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(GlobalServer).SysInfo(ctx, req.(*emptypb.Empty)) + } + return interceptor(ctx, in, info, handler) +} + +func _Global_Summary_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(emptypb.Empty) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(GlobalServer).Summary(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/m7s.Global/Summary", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(GlobalServer).Summary(ctx, req.(*emptypb.Empty)) + } + return interceptor(ctx, in, info, handler) +} + func _Global_Shutdown_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(RequestWithId) if err := dec(in); err != nil { @@ -224,6 +288,14 @@ var Global_ServiceDesc = grpc.ServiceDesc{ ServiceName: "m7s.Global", HandlerType: (*GlobalServer)(nil), Methods: []grpc.MethodDesc{ + { + MethodName: "SysInfo", + Handler: _Global_SysInfo_Handler, + }, + { + MethodName: "Summary", + Handler: _Global_Summary_Handler, + }, { MethodName: "Shutdown", Handler: _Global_Shutdown_Handler, diff --git a/pkg/util/index.go b/pkg/util/index.go index 6fd6272..d297209 100644 --- a/pkg/util/index.go +++ b/pkg/util/index.go @@ -1,5 +1,11 @@ package util +import ( + "log" + "os" + "path/filepath" +) + func Conditoinal[T any](cond bool, t, f T) T { if cond { return t @@ -19,3 +25,22 @@ func LenOfBuffers(b [][]byte) (n int) { } return } + +func initFatalLog() *os.File { + fatal_log_dir := "./fatal" + if _fatal_log := os.Getenv("M7S_FATAL_LOG"); _fatal_log != "" { + fatal_log_dir = _fatal_log + } + os.MkdirAll(fatal_log_dir, 0766) + fatal_log := filepath.Join(fatal_log_dir, "latest.log") + info, err := os.Stat(fatal_log) + if err == nil && info.Size() != 0 { + os.Rename(fatal_log, filepath.Join(fatal_log_dir, info.ModTime().Format("2006-01-02 15:04:05")+".log")) + } + logFile, err := os.OpenFile(fatal_log, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0666) + if err != nil { + log.Println("服务启动出错", "打开异常日志文件失败", err) + return nil + } + return logFile +} diff --git a/pkg/util/socket.go b/pkg/util/socket.go new file mode 100644 index 0000000..180f233 --- /dev/null +++ b/pkg/util/socket.go @@ -0,0 +1,272 @@ +package util + +import ( + "crypto/sha256" + "crypto/subtle" + "encoding/json" + "fmt" + "net" + "net/http" + "reflect" + "strconv" + "time" + + "gopkg.in/yaml.v3" +) + +func FetchValue[T any](t T) func() T { + return func() T { + return t + } +} + +const ( + APIErrorNone = 0 + APIErrorDecode = iota + 4000 + APIErrorQueryParse + APIErrorNoBody +) + +const ( + APIErrorNotFound = iota + 4040 + APIErrorNoStream + APIErrorNoConfig + APIErrorNoPusher + APIErrorNoSubscriber + APIErrorNoSEI +) + +const ( + APIErrorInternal = iota + 5000 + APIErrorJSONEncode + APIErrorPublish + APIErrorSave + APIErrorOpen +) + +type APIError struct { + Code int `json:"code"` + Message string `json:"msg"` +} + +type APIResult struct { + Code int `json:"code"` + Data any `json:"data"` + Message string `json:"msg"` +} + +func ReturnValue(v any, rw http.ResponseWriter, r *http.Request) { + ReturnFetchValue(FetchValue(v), rw, r) +} + +func ReturnOK(rw http.ResponseWriter, r *http.Request) { + ReturnError(0, "ok", rw, r) +} + +func ReturnError(code int, msg string, rw http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + isJson := query.Get("format") == "json" + if isJson { + if err := json.NewEncoder(rw).Encode(APIError{code, msg}); err != nil { + json.NewEncoder(rw).Encode(APIError{ + Code: APIErrorJSONEncode, + Message: err.Error(), + }) + } + } else { + switch true { + case code == 0: + http.Error(rw, msg, http.StatusOK) + case code/10 == 404: + http.Error(rw, msg, http.StatusNotFound) + case code > 5000: + http.Error(rw, msg, http.StatusInternalServerError) + default: + http.Error(rw, msg, http.StatusBadRequest) + } + } +} +func ReturnFetchList[T any](fetch func() []T, rw http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + isYaml := query.Get("format") == "yaml" + isJson := query.Get("format") == "json" + pageSize := query.Get("pageSize") + pageNum := query.Get("pageNum") + data := fetch() + var output any + output = data + if pageSize != "" && pageNum != "" { + pageSizeInt, _ := strconv.Atoi(pageSize) + pageNumInt, _ := strconv.Atoi(pageNum) + if pageSizeInt > 0 && pageNumInt > 0 { + start := (pageNumInt - 1) * pageSizeInt + end := pageNumInt * pageSizeInt + if start > len(data) { + start = len(data) + } + if end > len(data) { + end = len(data) + } + output = map[string]any{ + "total": len(data), + "list": data[start:end], + "pageSize": pageSizeInt, + "pageNum": pageNumInt, + } + } + } + rw.Header().Set("Content-Type", Conditoinal(isYaml, "text/yaml", "application/json")) + if isYaml { + if err := yaml.NewEncoder(rw).Encode(output); err != nil { + http.Error(rw, err.Error(), http.StatusInternalServerError) + } + } else if isJson { + if err := json.NewEncoder(rw).Encode(APIResult{ + Code: 0, + Data: output, + Message: "ok", + }); err != nil { + json.NewEncoder(rw).Encode(APIError{ + Code: APIErrorJSONEncode, + Message: err.Error(), + }) + } + } else { + if err := json.NewEncoder(rw).Encode(output); err != nil { + http.Error(rw, err.Error(), http.StatusInternalServerError) + } + } +} +func ReturnFetchValue[T any](fetch func() T, rw http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + isYaml := query.Get("format") == "yaml" + isJson := query.Get("format") == "json" + tickDur, err := time.ParseDuration(query.Get("interval")) + if err != nil { + tickDur = time.Second + } + if r.Header.Get("Accept") == "text/event-stream" { + sse := NewSSE(rw, r.Context()) + tick := time.NewTicker(tickDur) + defer tick.Stop() + writer := Conditoinal(isYaml, sse.WriteYAML, sse.WriteJSON) + writer(fetch()) + for range tick.C { + if writer(fetch()) != nil { + return + } + } + } else { + data := fetch() + rw.Header().Set("Content-Type", Conditoinal(isYaml, "text/yaml", "application/json")) + if isYaml { + if err := yaml.NewEncoder(rw).Encode(data); err != nil { + http.Error(rw, err.Error(), http.StatusInternalServerError) + } + } else if isJson { + if err := json.NewEncoder(rw).Encode(APIResult{ + Code: 0, + Data: data, + Message: "ok", + }); err != nil { + json.NewEncoder(rw).Encode(APIError{ + Code: APIErrorJSONEncode, + Message: err.Error(), + }) + } + } else { + t := reflect.TypeOf(data) + switch t.Kind() { + case reflect.String, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Float32, reflect.Float64: + rw.Header().Set("Content-Type", "text/plain") + fmt.Fprint(rw, data) + default: + if err := json.NewEncoder(rw).Encode(data); err != nil { + http.Error(rw, err.Error(), http.StatusInternalServerError) + } + } + } + } +} + +func ListenUDP(address string, networkBuffer int) (*net.UDPConn, error) { + addr, err := net.ResolveUDPAddr("udp", address) + if err != nil { + return nil, err + } + conn, err := net.ListenUDP("udp", addr) + if err != nil { + return nil, err + } + if err = conn.SetReadBuffer(networkBuffer); err != nil { + return nil, err + } + if err = conn.SetWriteBuffer(networkBuffer); err != nil { + return nil, err + } + return conn, err +} + +// CORS 加入跨域策略头包含CORP +func CORS(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + header := w.Header() + header.Set("Access-Control-Allow-Credentials", "true") + header.Set("Cross-Origin-Resource-Policy", "cross-origin") + header.Set("Access-Control-Allow-Headers", "Content-Type,Access-Token") + origin := r.Header["Origin"] + if len(origin) == 0 { + header.Set("Access-Control-Allow-Origin", "*") + } else { + header.Set("Access-Control-Allow-Origin", origin[0]) + } + if next != nil && r.Method != "OPTIONS" { + next.ServeHTTP(w, r) + } + }) +} + +func BasicAuth(u, p string, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Extract the username and password from the request + // Authorization header. If no Authentication header is present + // or the header value is invalid, then the 'ok' return value + // will be false. + username, password, ok := r.BasicAuth() + if ok { + // Calculate SHA-256 hashes for the provided and expected + // usernames and passwords. + usernameHash := sha256.Sum256([]byte(username)) + passwordHash := sha256.Sum256([]byte(password)) + expectedUsernameHash := sha256.Sum256([]byte(u)) + expectedPasswordHash := sha256.Sum256([]byte(p)) + + // 使用 subtle.ConstantTimeCompare() 进行校验 + // the provided username and password hashes equal the + // expected username and password hashes. ConstantTimeCompare + // 如果值相等,则返回1,否则返回0。 + // Importantly, we should to do the work to evaluate both the + // username and password before checking the return values to + // 避免泄露信息。 + usernameMatch := (subtle.ConstantTimeCompare(usernameHash[:], expectedUsernameHash[:]) == 1) + passwordMatch := (subtle.ConstantTimeCompare(passwordHash[:], expectedPasswordHash[:]) == 1) + + // If the username and password are correct, then call + // the next handler in the chain. Make sure to return + // afterwards, so that none of the code below is run. + if usernameMatch && passwordMatch { + if next != nil { + next.ServeHTTP(w, r) + } + return + } + } + + // If the Authentication header is not present, is invalid, or the + // username or password is wrong, then set a WWW-Authenticate + // header to inform the client that we expect them to use basic + // authentication and send a 401 Unauthorized response. + w.Header().Set("WWW-Authenticate", `Basic realm="restricted", charset="UTF-8"`) + http.Error(w, "Unauthorized", http.StatusUnauthorized) + }) +} diff --git a/pkg/util/sse.go b/pkg/util/sse.go index fb97be7..3b03785 100644 --- a/pkg/util/sse.go +++ b/pkg/util/sse.go @@ -52,6 +52,10 @@ func NewSSE(w http.ResponseWriter, ctx context.Context) *SSE { header.Set("Connection", "keep-alive") header.Set("X-Accel-Buffering", "no") header.Set("Access-Control-Allow-Origin", "*") + // rw.Header().Set("Access-Control-Allow-Headers", "Content-Type,Authorization") + // rw.Header().Set("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS") + // rw.Header().Set("Access-Control-Allow-Credentials", "true") + // rw.Header().Set("Transfer-Encoding", "chunked") return &SSE{ ResponseWriter: w, Context: ctx, diff --git a/pkg/util/stderr.go b/pkg/util/stderr.go new file mode 100644 index 0000000..a01b8fe --- /dev/null +++ b/pkg/util/stderr.go @@ -0,0 +1,16 @@ +//go:build (linux && !arm64) || darwin + +package util + +import ( + "os" + "syscall" +) + +func init() { + logFile := initFatalLog() + if logFile != nil { + // 将进程标准出错重定向至文件,进程崩溃时运行时将向该文件记录协程调用栈信息 + syscall.Dup2(int(logFile.Fd()), int(os.Stderr.Fd())) + } +} diff --git a/pkg/util/stderr_arm64.go b/pkg/util/stderr_arm64.go new file mode 100644 index 0000000..292ad29 --- /dev/null +++ b/pkg/util/stderr_arm64.go @@ -0,0 +1,16 @@ +//go:build linux && !darwin + +package util + +import ( + "os" + "syscall" +) + +func init() { + logFile := initFatalLog() + if logFile != nil { + // 将进程标准出错重定向至文件,进程崩溃时运行时将向该文件记录协程调用栈信息 + syscall.Dup3(int(logFile.Fd()), int(os.Stderr.Fd()), 0) + } +} diff --git a/pkg/util/stderr_windows.go b/pkg/util/stderr_windows.go new file mode 100644 index 0000000..f775188 --- /dev/null +++ b/pkg/util/stderr_windows.go @@ -0,0 +1,38 @@ +//go:build windows + +package util + +import ( + "log" + "os" + "syscall" +) + +var ( + kernel32 = syscall.MustLoadDLL("kernel32.dll") + procSetStdHandle = kernel32.MustFindProc("SetStdHandle") +) + +func setStdHandle(stdhandle int32, handle syscall.Handle) error { + r0, _, e1 := syscall.Syscall(procSetStdHandle.Addr(), 2, uintptr(stdhandle), uintptr(handle), 0) + if r0 == 0 { + if e1 != 0 { + return error(e1) + } + return syscall.EINVAL + } + return nil +} + +// redirectStderr to the file passed in +func init() { + logFile := initFatalLog() + if logFile != nil { + err := setStdHandle(syscall.STD_ERROR_HANDLE, syscall.Handle(logFile.Fd())) + if err != nil { + log.Fatalf("Failed to redirect stderr to file: %v", err) + } + // SetStdHandle does not affect prior references to stderr + os.Stderr = logFile + } +} diff --git a/plugin.go b/plugin.go index 9196738..133cd03 100644 --- a/plugin.go +++ b/plugin.go @@ -40,6 +40,7 @@ func (plugin *PluginMeta) Init(s *Server, userConfig map[string]any) { p.Meta = plugin p.server = s p.Logger = s.Logger.With("plugin", plugin.Name) + p.Context, p.CancelCauseFunc = context.WithCancelCause(s.Context) if os.Getenv(strings.ToUpper(plugin.Name)+"_ENABLE") == "false" { p.Disabled = true p.Warn("disabled by env") @@ -85,7 +86,7 @@ func (plugin *PluginMeta) Init(s *Server, userConfig map[string]any) { if plugin.ServiceDesc != nil && s.grpcServer != nil { s.grpcServer.RegisterService(plugin.ServiceDesc, instance) if plugin.RegisterGRPCHandler != nil { - err = plugin.RegisterGRPCHandler(s.Context, s.config.HTTP.GetGRPCMux(), s.grpcClientConn) + err = plugin.RegisterGRPCHandler(p.Context, s.config.HTTP.GetGRPCMux(), s.grpcClientConn) if err != nil { p.Error("init", "error", err) p.Stop(err) @@ -155,6 +156,10 @@ func (Plugin) nothing() { } +func (p *Plugin) GetGlobalCommonConf() *config.Common { + return p.server.GetCommonConf() +} + func (p *Plugin) GetCommonConf() *config.Common { return &p.config } @@ -184,7 +189,6 @@ func (p *Plugin) Stop(err error) { } func (p *Plugin) Start() { - p.Context, p.CancelCauseFunc = context.WithCancelCause(p.server.Context) httpConf := &p.config.HTTP if httpConf.ListenAddrTLS != "" && (httpConf.ListenAddrTLS != p.server.config.HTTP.ListenAddrTLS) { p.Info("https listen at ", "addr", httpConf.ListenAddrTLS) diff --git a/plugin/console/index.go b/plugin/console/index.go new file mode 100644 index 0000000..e99c3bb --- /dev/null +++ b/plugin/console/index.go @@ -0,0 +1,154 @@ +package plugin_console + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" + + "github.com/quic-go/quic-go" + "m7s.live/m7s/v5" +) + +type myResponseWriter struct { +} + +func (*myResponseWriter) Header() http.Header { + return make(http.Header) +} +func (*myResponseWriter) WriteHeader(statusCode int) { +} +func (w *myResponseWriter) Flush() { +} + +type myResponseWriter2 struct { + quic.Stream + myResponseWriter +} + +type myResponseWriter3 struct { + handshake bool + myResponseWriter2 + quic.Connection +} + +func (w *myResponseWriter3) Write(b []byte) (int, error) { + if !w.handshake { + w.handshake = true + return len(b), nil + } + println(string(b)) + return w.Stream.Write(b) +} + +func (w *myResponseWriter3) Hijack() (net.Conn, *bufio.ReadWriter, error) { + return net.Conn(w), bufio.NewReadWriter(bufio.NewReader(w), bufio.NewWriter(w)), nil +} + +type ConsolePlugin struct { + m7s.Plugin + Server string `default:"console.monibuca.com:44944" desc:"远程控制台地址"` //远程控制台地址 + Secret string `desc:"远程控制台密钥"` //远程控制台密钥 +} + +var _ = m7s.InstallPlugin[ConsolePlugin]() + +func (cfg *ConsolePlugin) OnInit() error { + tlsConf := &tls.Config{ + InsecureSkipVerify: true, + NextProtos: []string{"monibuca"}, + } + conn, err := quic.DialAddr(cfg.Context, cfg.Server, tlsConf, &quic.Config{ + KeepAlivePeriod: time.Second * 10, + EnableDatagrams: true, + }) + if stream := quic.Stream(nil); err == nil { + if stream, err = conn.OpenStreamSync(cfg.Context); err == nil { + _, err = stream.Write(append([]byte{1}, (cfg.Secret + "\n")...)) + if msg := []byte(nil); err == nil { + if msg, err = bufio.NewReader(stream).ReadSlice(0); err == nil { + var rMessage map[string]any + if err = json.Unmarshal(msg[:len(msg)-1], &rMessage); err == nil { + if rMessage["code"].(float64) != 0 { + // cfg.Error("response from console server ", cfg.Server, rMessage["msg"]) + return fmt.Errorf("response from console server %s %s", cfg.Server, rMessage["msg"]) + } else { + // cfg.reportStream = stream + cfg.Info("response from console server ", cfg.Server, rMessage) + // if v, ok := rMessage["enableReport"]; ok { + // cfg.enableReport = v.(bool) + // } + // if v, ok := rMessage["instanceId"]; ok { + // cfg.instanceId = v.(string) + // } + } + } + } + } + } + } + go func() { + for err == nil { + var s quic.Stream + if s, err = conn.AcceptStream(cfg.Context); err == nil { + go cfg.ReceiveRequest(s, conn) + } + } + }() + return err +} + +func (cfg *ConsolePlugin) ReceiveRequest(s quic.Stream, conn quic.Connection) error { + defer s.Close() + wr := &myResponseWriter2{Stream: s} + reader := bufio.NewReader(s) + var req *http.Request + url, _, err := reader.ReadLine() + if err == nil { + ctx, cancel := context.WithCancel(s.Context()) + defer cancel() + req, err = http.NewRequestWithContext(ctx, "GET", string(url), reader) + for err == nil { + var h []byte + if h, _, err = reader.ReadLine(); len(h) > 0 { + if b, a, f := strings.Cut(string(h), ": "); f { + req.Header.Set(b, a) + } + } else { + break + } + } + + if err == nil { + h := cfg.GetGlobalCommonConf().GetHandler() + if req.Header.Get("Accept") == "text/event-stream" { + go h.ServeHTTP(wr, req) + } else if req.Header.Get("Upgrade") == "websocket" { + var writer myResponseWriter3 + writer.Stream = s + writer.Connection = conn + req.Host = req.Header.Get("Host") + if req.Host == "" { + req.Host = req.URL.Host + } + if req.Host == "" { + req.Host = "m7s.live" + } + h.ServeHTTP(&writer, req) //建立websocket连接,握手 + } else { + h.ServeHTTP(wr, req) + } + } + io.ReadAll(s) + } + if err != nil { + cfg.Error("read console server", "err", err) + } + return err +} diff --git a/server.go b/server.go index 442dc93..7c8c86e 100644 --- a/server.go +++ b/server.go @@ -2,7 +2,6 @@ package m7s import ( "context" - "errors" "fmt" "log/slog" "net" @@ -45,20 +44,22 @@ type Server struct { pb.UnimplementedGlobalServer Plugin config.Engine - ID int - eventChan chan any - Plugins []*Plugin - Streams util.Collection[string, *Publisher] - Pulls util.Collection[string, *Puller] - Pushs util.Collection[string, *Pusher] - Waiting map[string][]*Subscriber - Subscribers util.Collection[int, *Subscriber] - LogHandler MultiLogHandler - pidG int - sidG int - apiList []string - grpcServer *grpc.Server - grpcClientConn *grpc.ClientConn + ID int + eventChan chan any + Plugins []*Plugin + Streams util.Collection[string, *Publisher] + Pulls util.Collection[string, *Puller] + Pushs util.Collection[string, *Pusher] + Waiting map[string][]*Subscriber + Subscribers util.Collection[int, *Subscriber] + LogHandler MultiLogHandler + pidG int + sidG int + apiList []string + grpcServer *grpc.Server + grpcClientConn *grpc.ClientConn + lastSummaryTime time.Time + lastSummary *pb.SummaryResponse } func NewServer() (s *Server) { @@ -103,6 +104,7 @@ func (s *Server) reset() { } func (s *Server) Run(ctx context.Context, conf any) (err error) { + s.StartTime = time.Now() for err = s.run(ctx, conf); err == ErrRestart; err = s.run(ctx, conf) { s.reset() } @@ -265,6 +267,13 @@ func (s *Server) eventLoop() { switch v := event.(type) { case *util.Promise[any]: switch vv := v.Value.(type) { + case func(): + vv() + v.Fulfill(nil) + continue + case func() error: + v.Fulfill(vv()) + continue case *Publisher: err := s.OnPublish(vv) if v.Fulfill(err); err != nil { @@ -310,30 +319,6 @@ func (s *Server) eventLoop() { s.Pushs.Add(vv) event = v.Value } - case *pb.StreamSnapRequest: - if pub, ok := s.Streams.Get(vv.StreamPath); ok { - v.Resolve(pub) - } else { - v.Fulfill(ErrNotFound) - } - continue - case *pb.StopSubscribeRequest: - if subscriber, ok := s.Subscribers.Get(int(vv.Id)); ok { - subscriber.Stop(errors.New("stop by api")) - v.Fulfill(nil) - } else { - v.Fulfill(ErrNotFound) - } - continue - case *pb.StreamListRequest: - var streams []*pb.StreamInfo - for _, publisher := range s.Streams.Items { - streams = append(streams, &pb.StreamInfo{ - Path: publisher.StreamPath, - }) - } - v.Resolve(&pb.StreamListResponse{List: streams}) - continue } case slog.Handler: s.LogHandler.Add(v)