apps/monibuca
1
0
Fork 0
mirror of https://github.com/langhuihui/monibuca.git synced 2026-05-07 18:51:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add more info to sysinfo

Browse Source
This commit is contained in:
langhuihui
2024-05-22 17:25:28 +08:00
parent ac312d5160
commit a88ce033fc
7 changed files with 770 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api.go
+27
View File
@@ -6,6 +6,7 @@ import (
"errors"
"net"
"net/http"
"runtime"
"strings"
"time"
@@ -35,6 +36,17 @@ func (s *Server) SysInfo(context.Context, *emptypb.Empty) (res *pb.SysInfoRespon
Version: Version,
LocalIP: localIP,
StartTime: timestamppb.New(s.StartTime),
GoVersion: runtime.Version(),
Os: runtime.GOOS,
Arch: runtime.GOARCH,
Cpus: int32(runtime.NumCPU()),
}
for _, p := range s.Plugins.Items {
res.Plugins = append(res.Plugins, &pb.PluginInfo{
Name: p.Meta.Name,
Version: p.Meta.Version,
Enable: !p.Disabled,
})
}
return
}
@@ -258,6 +270,18 @@ func (s *Server) StreamList(_ context.Context, req *pb.StreamListRequest) (res *
return
}
func (s *Server) WaitList(context.Context, *emptypb.Empty) (res *pb.StreamWaitListResponse, err error) {
s.Call(func() {
res = &pb.StreamWaitListResponse{
List: make(map[string]int32),
}
for streamPath, subs := range s.Waiting {
res.List[streamPath] = int32(len(subs))
}
})
return
}
func (s *Server) API_Summary_SSE(rw http.ResponseWriter, r *http.Request) {
util.ReturnFetchValue(func() *pb.SummaryResponse {
ret, _ := s.Summary(r.Context(), nil)
@@ -306,6 +330,9 @@ func (s *Server) Summary(context.Context, *emptypb.Empty) (res *pb.SummaryRespon
netWorks = append(netWorks, info)
}
res.StreamCount = int32(s.Streams.Length)
res.PullCount = int32(s.Pulls.Length)
res.PushCount = int32(s.Pushs.Length)
res.SubscribeCount = int32(s.Subscribers.Length)
res.NetWork = netWorks
s.lastSummary = res
s.lastSummaryTime = time.Now()
example/multiple/config2.yaml
+2
View File
@@ -4,6 +4,8 @@ global:
listenaddr: :8081
listenaddrtls: :8555
disableall: true
console:
secret: 00aea3af031f134d6307618b05ec4899
rtmp:
enable: true
chunksize: 2048
pb/global.pb.go
+610 -370
View File
File diff suppressed because it is too large Load Diff
pb/global.pb.gw.go
+69
View File
@@ -208,6 +208,24 @@ func local_request_Global_StreamList_0(ctx context.Context, marshaler runtime.Ma
}
func request_Global_WaitList_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.WaitList(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
return msg, metadata, err
}
func local_request_Global_WaitList_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.WaitList(ctx, &protoReq)
return msg, metadata, err
}
func request_Global_StreamInfo_0(ctx context.Context, marshaler runtime.Marshaler, client GlobalClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
var protoReq StreamSnapRequest
var metadata runtime.ServerMetadata
@@ -789,6 +807,31 @@ func RegisterGlobalHandlerServer(ctx context.Context, mux *runtime.ServeMux, ser
})
mux.Handle("GET", pattern_Global_WaitList_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/WaitList", runtime.WithHTTPPathPattern("/api/stream/waitlist"))
if err != nil {
runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
return
}
resp, md, err := local_request_Global_WaitList_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_WaitList_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
})
mux.Handle("GET", pattern_Global_StreamInfo_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
ctx, cancel := context.WithCancel(req.Context())
defer cancel()
@@ -1140,6 +1183,28 @@ func RegisterGlobalHandlerClient(ctx context.Context, mux *runtime.ServeMux, cli
})
mux.Handle("GET", pattern_Global_WaitList_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/WaitList", runtime.WithHTTPPathPattern("/api/stream/waitlist"))
if err != nil {
runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
return
}
resp, md, err := request_Global_WaitList_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_WaitList_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
})
mux.Handle("GET", pattern_Global_StreamInfo_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
ctx, cancel := context.WithCancel(req.Context())
defer cancel()
@@ -1330,6 +1395,8 @@ var (
pattern_Global_StreamList_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"api", "stream", "list"}, ""))
pattern_Global_WaitList_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"api", "stream", "waitlist"}, ""))
pattern_Global_StreamInfo_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 3, 0, 4, 1, 5, 3}, []string{"api", "stream", "info", "streamPath"}, ""))
pattern_Global_GetSubscribers_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 3, 0, 4, 1, 5, 2}, []string{"api", "subscribers", "streamPath"}, ""))
@@ -1358,6 +1425,8 @@ var (
forward_Global_StreamList_0 = runtime.ForwardResponseMessage
forward_Global_WaitList_0 = runtime.ForwardResponseMessage
forward_Global_StreamInfo_0 = runtime.ForwardResponseMessage
forward_Global_GetSubscribers_0 = runtime.ForwardResponseMessage
pb/global.proto
+23
View File
@@ -32,6 +32,11 @@ service Global {
get: "/api/stream/list"
};
}
rpc WaitList (google.protobuf.Empty) returns (StreamWaitListResponse) {
option (google.api.http) = {
get: "/api/stream/waitlist"
};
}
rpc StreamInfo (StreamSnapRequest) returns (StreamInfoResponse) {
option (google.api.http) = {
get: "/api/stream/info/{streamPath=**}"
@@ -125,12 +130,26 @@ message SummaryResponse {
Usage hardDisk = 4;
repeated NetWorkInfo netWork = 5;
int32 streamCount = 6;
int32 subscribeCount = 7;
int32 pullCount = 8;
int32 pushCount = 9;
}
message PluginInfo {
string name = 1;
string version = 2;
bool enable = 3;
}
message SysInfoResponse {
google.protobuf.Timestamp startTime = 1;
string localIP = 2;
string version = 3;
string goVersion = 4;
string os = 5;
string arch = 6;
int32 cpus = 7;
repeated PluginInfo plugins = 8;
}
message StreamListRequest {
@@ -145,6 +164,10 @@ message StreamListResponse {
repeated StreamInfoResponse list = 4;
}
message StreamWaitListResponse {
map<string, int32> list = 1;
}
message StreamSnapRequest {
string streamPath = 1;
}
pb/global_grpc.pb.go
+36
View File
@@ -28,6 +28,7 @@ type GlobalClient interface {
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)
WaitList(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*StreamWaitListResponse, error)
StreamInfo(ctx context.Context, in *StreamSnapRequest, opts ...grpc.CallOption) (*StreamInfoResponse, error)
GetSubscribers(ctx context.Context, in *SubscribersRequest, opts ...grpc.CallOption) (*SubscribersResponse, error)
AudioTrackSnap(ctx context.Context, in *StreamSnapRequest, opts ...grpc.CallOption) (*TrackSnapShotResponse, error)
@@ -91,6 +92,15 @@ func (c *globalClient) StreamList(ctx context.Context, in *StreamListRequest, op
return out, nil
}
func (c *globalClient) WaitList(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*StreamWaitListResponse, error) {
out := new(StreamWaitListResponse)
err := c.cc.Invoke(ctx, "/m7s.Global/WaitList", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *globalClient) StreamInfo(ctx context.Context, in *StreamSnapRequest, opts ...grpc.CallOption) (*StreamInfoResponse, error) {
out := new(StreamInfoResponse)
err := c.cc.Invoke(ctx, "/m7s.Global/StreamInfo", in, out, opts...)
@@ -172,6 +182,7 @@ type GlobalServer interface {
Shutdown(context.Context, *RequestWithId) (*emptypb.Empty, error)
Restart(context.Context, *RequestWithId) (*emptypb.Empty, error)
StreamList(context.Context, *StreamListRequest) (*StreamListResponse, error)
WaitList(context.Context, *emptypb.Empty) (*StreamWaitListResponse, error)
StreamInfo(context.Context, *StreamSnapRequest) (*StreamInfoResponse, error)
GetSubscribers(context.Context, *SubscribersRequest) (*SubscribersResponse, error)
AudioTrackSnap(context.Context, *StreamSnapRequest) (*TrackSnapShotResponse, error)
@@ -202,6 +213,9 @@ func (UnimplementedGlobalServer) Restart(context.Context, *RequestWithId) (*empt
func (UnimplementedGlobalServer) StreamList(context.Context, *StreamListRequest) (*StreamListResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method StreamList not implemented")
}
func (UnimplementedGlobalServer) WaitList(context.Context, *emptypb.Empty) (*StreamWaitListResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method WaitList not implemented")
}
func (UnimplementedGlobalServer) StreamInfo(context.Context, *StreamSnapRequest) (*StreamInfoResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method StreamInfo not implemented")
}
@@ -329,6 +343,24 @@ func _Global_StreamList_Handler(srv interface{}, ctx context.Context, dec func(i
return interceptor(ctx, in, info, handler)
}
func _Global_WaitList_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).WaitList(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/m7s.Global/WaitList",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(GlobalServer).WaitList(ctx, req.(*emptypb.Empty))
}
return interceptor(ctx, in, info, handler)
}
func _Global_StreamInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(StreamSnapRequest)
if err := dec(in); err != nil {
@@ -500,6 +532,10 @@ var Global_ServiceDesc = grpc.ServiceDesc{
MethodName: "StreamList",
Handler: _Global_StreamList_Handler,
},
{
MethodName: "WaitList",
Handler: _Global_WaitList_Handler,
},
{
MethodName: "StreamInfo",
Handler: _Global_StreamInfo_Handler,
server.go
+3 -1
View File
@@ -105,6 +105,8 @@ func (s *Server) reset() {
server.LogHandler = MultiLogHandler{}
server.LogHandler.SetLevel(slog.LevelInfo)
server.LogHandler.Add(console.NewHandler(os.Stdout, nil))
server.OnAuthPubs = s.OnAuthPubs
server.OnAuthSubs = s.OnAuthSubs
// server.Logger = slog.New(&server.LogHandler).With("server", s.ID)
*s = server
}
@@ -466,7 +468,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.ServeFile(w, r, "favicon.ico")
return
}
fmt.Fprintf(w, "Monibuca Engine %s StartTime:%s\n", Version, s.StartTime)
fmt.Fprintf(w, "visit:%s\nMonibuca Engine %s StartTime:%s\n", r.URL.Path, Version, s.StartTime)
for _, plugin := range s.Plugins.Items {
fmt.Fprintf(w, "Plugin %s Version:%s\n", plugin.Meta.Name, plugin.Meta.Version)
}