diff --git a/api/courses.go b/api/courses.go index 840256a62..69b85c56c 100644 --- a/api/courses.go +++ b/api/courses.go @@ -8,9 +8,8 @@ import ( "fmt" "io" "net/http" - "net/http/httputil" - "net/url" "os" + "path/filepath" "regexp" "sort" "strconv" @@ -28,10 +27,15 @@ import ( "github.com/TUM-Dev/gocast/model" "github.com/TUM-Dev/gocast/tools" "github.com/TUM-Dev/gocast/tools/tum" + + "github.com/tum-dev/gocast/runner/protobuf" ) -func configGinCourseRouter(router *gin.Engine, daoWrapper dao.DaoWrapper) { - routes := coursesRoutes{daoWrapper} +func configGinCourseRouter(router *gin.Engine, daoWrapper dao.DaoWrapper, manager runnerManager) { + routes := coursesRoutes{ + DaoWrapper: daoWrapper, + manager: manager, + } router.POST("/api/course/activate/:token", routes.activateCourseByToken) router.GET("/api/lecture-halls-by-id", routes.lectureHallsByID) @@ -105,6 +109,11 @@ func configGinCourseRouter(router *gin.Engine, daoWrapper dao.DaoWrapper) { type coursesRoutes struct { dao.DaoWrapper + manager runnerManager +} + +type runnerManager interface { + SendVODJob(ctx context.Context, streamID uint, version protobuf.StreamVersion, recordingDir string) error } const ( @@ -466,44 +475,87 @@ func (r coursesRoutes) uploadVODMedia(c *gin.Context) { return } - key := uuid.NewV4().String() - err = r.UploadKeyDao.CreateUploadKey(key, stream.ID, req.VideoType) + // Get the uploaded file + file, header, err := c.Request.FormFile("file") + if err != nil { + _ = c.Error(tools.RequestError{ + Status: http.StatusBadRequest, + CustomMessage: "can not read uploaded file", + Err: err, + }) + return + } + defer file.Close() + + // Create directory structure in Ceph: mass/streamID/videoType/ + streamDir := filepath.Join(tools.Cfg.Paths.Mass, fmt.Sprintf("%d", stream.ID), string(req.VideoType)) + err = os.MkdirAll(streamDir, os.ModePerm) + if err != nil { + _ = c.Error(tools.RequestError{ + Status: http.StatusInternalServerError, + CustomMessage: "can not create storage directory", + Err: err, + }) + return + } + + // Save file to Ceph + destPath := filepath.Join(streamDir, header.Filename) + destFile, err := os.Create(destPath) if err != nil { _ = c.Error(tools.RequestError{ Status: http.StatusInternalServerError, - CustomMessage: "can not create upload key", + CustomMessage: "can not create destination file", Err: err, }) return } - workers := r.WorkerDao.GetAliveWorkers() - if len(workers) == 0 { + defer destFile.Close() + + _, err = io.Copy(destFile, file) + if err != nil { _ = c.Error(tools.RequestError{ Status: http.StatusInternalServerError, - CustomMessage: "no workers available", + CustomMessage: "can not save file", Err: err, }) return } - w := workers[getWorkerWithLeastWorkload(workers)] - u, err := url.Parse("http://" + w.Host + ":" + WorkerHTTPPort + "/upload?" + c.Request.URL.Query().Encode() + "&key=" + key) + + logger.Info("File saved to Ceph", "path", destPath, "streamID", stream.ID, "videoType", req.VideoType) + + // Send job to runner to process the VOD + err = r.sendVODJobToRunner(c.Request.Context(), stream.ID, req.VideoType, streamDir) if err != nil { + logger.Error("Failed to send VOD job to runner", "err", err, "streamID", stream.ID) _ = c.Error(tools.RequestError{ Status: http.StatusInternalServerError, - CustomMessage: fmt.Sprintf("parse proxy url: %v", err), + CustomMessage: "file uploaded but failed to start processing", Err: err, }) return } - p := httputil.NewSingleHostReverseProxy(u) - p.Director = func(req *http.Request) { - req.URL.Scheme = u.Scheme - req.URL.Host = u.Host - req.Host = u.Host - req.URL.Path = u.Path - req.URL.RawQuery = u.RawQuery + + c.JSON(http.StatusOK, gin.H{"message": "file uploaded successfully, processing started"}) +} + +// sendVODJobToRunner sends a VOD processing job to an available runner +func (r coursesRoutes) sendVODJobToRunner(ctx context.Context, streamID uint, videoType model.VideoType, recordingDir string) error { + // Convert VideoType to StreamVersion + var version protobuf.StreamVersion + switch videoType { + case model.VideoTypeCombined: + version = protobuf.StreamVersion_STREAM_VERSION_COMBINED + case model.VideoTypePresentation: + version = protobuf.StreamVersion_STREAM_VERSION_PRESENTATION + case model.VideoTypeCamera: + version = protobuf.StreamVersion_STREAM_VERSION_CAMERA + default: + return fmt.Errorf("unsupported video type: %s", videoType) } - p.ServeHTTP(c.Writer, c.Request) + + // Send job to runner via manager + return r.manager.SendVODJob(ctx, streamID, version, recordingDir) } // updateSourceSettings updates the CameraPresets of a course diff --git a/api/router.go b/api/router.go index 707988705..7128c1674 100755 --- a/api/router.go +++ b/api/router.go @@ -38,7 +38,7 @@ func ConfigGinRouter( configGinStreamRestRouter(router, daoWrapper) configGinUsersRouter(router, daoWrapper) - configGinCourseRouter(router, daoWrapper) + configGinCourseRouter(router, daoWrapper, manager) configGinDownloadRouter(router, daoWrapper) configGinDownloadICSRouter(router, daoWrapper) configGinLectureHallApiRouter(router, daoWrapper, camService, tools.Cfg.Paths.Static) diff --git a/dao/upload_key.go b/dao/upload_key.go index 080ef7f6d..dd5fcadeb 100644 --- a/dao/upload_key.go +++ b/dao/upload_key.go @@ -8,6 +8,7 @@ import ( //go:generate go tool mockgen -source=upload_key.go -destination ../mock_dao/upload_key.go +// deprecated: delete with worker type UploadKeyDao interface { GetUploadKey(key string) (model.UploadKey, error) CreateUploadKey(key string, stream uint, videoType model.VideoType) error diff --git a/model/upload-key.go b/model/upload-key.go index 7e919ed95..884c9cac6 100644 --- a/model/upload-key.go +++ b/model/upload-key.go @@ -14,6 +14,7 @@ func (v VideoType) Valid() bool { return v == VideoTypeCombined || v == VideoTypePresentation || v == VideoTypeCamera } +// deprecated: delete with worker // UploadKey represents a key that is created when a user uploads a file, // sent to the worker with the upload request and back to TUM-Live to authenticate the request. type UploadKey struct { diff --git a/pkg/runner_manager/manager.go b/pkg/runner_manager/manager.go index 100547074..ab4a8241c 100644 --- a/pkg/runner_manager/manager.go +++ b/pkg/runner_manager/manager.go @@ -237,6 +237,27 @@ func (m *Manager) getClient(ctx context.Context) (protobuf.RunnerServiceClient, return protobuf.NewRunnerServiceClient(conn), nil } +// SendVODJob sends a VOD processing job to an available runner +func (m *Manager) SendVODJob(ctx context.Context, streamID uint, version protobuf.StreamVersion, recordingDir string) error { + client, err := m.getClient(ctx) + if err != nil { + return fmt.Errorf("get runner client: %w", err) + } + + streamIDUint64 := uint64(streamID) + _, err = client.HandleVOD(ctx, &protobuf.HandleVODRequest{ + StreamId: &streamIDUint64, + Version: &version, + Filepath: &recordingDir, + }) + if err != nil { + return fmt.Errorf("send HandleVOD request: %w", err) + } + + m.logger.Info("VOD job sent to runner", "streamID", streamID, "version", version) + return nil +} + func (m *Manager) streamStarted(ctx context.Context, req *protobuf.StreamStartNotification) error { // This is usually called in bursts, which introduces a chance for race conditions, // where a stream is fetched and overwrites the url that the other requests added. diff --git a/runner/handlers.go b/runner/handlers.go index e2d5e063e..83cd99f10 100644 --- a/runner/handlers.go +++ b/runner/handlers.go @@ -45,3 +45,23 @@ func (r *Runner) RequestStreamEnd(_ context.Context, req *protobuf.StreamEndRequ } return nil, status.Errorf(codes.NotFound, "stream not found") } + +func (r *Runner) HandleVOD(_ context.Context, req *protobuf.HandleVODRequest) (*protobuf.HandleVODResponse, error) { + data := map[string]any{ + "streamID": req.GetStreamId(), + "streamVersion": req.GetVersion().String(), + "recordingDir": req.GetFilepath(), + } + r.log.Info("HandleVOD data constructed", "data", data) + a := []actions.Action{ + actions.CheckCodec, + actions.MkVOD, + actions.CheckVoD, + actions.MkThumb, + } + + jID := r.RunAction(a, data, r.log.With("stream_id", req.GetStreamId(), "stream_version", req.GetVersion(), "input", req.GetFilepath())) + r.log.Info("job added", "ID", jID) + + return &protobuf.HandleVODResponse{JobId: ptr.Take(jID)}, nil +} diff --git a/runner/pkg/actions/checkcodec.go b/runner/pkg/actions/checkcodec.go new file mode 100644 index 000000000..99b0feff4 --- /dev/null +++ b/runner/pkg/actions/checkcodec.go @@ -0,0 +1,58 @@ +package actions + +import ( + "context" + "fmt" + "log/slog" + "path" + + "github.com/tum-dev/gocast/runner/pkg/ffmpeg" + "github.com/tum-dev/gocast/runner/pkg/metrics" + "github.com/tum-dev/gocast/runner/protobuf" +) + +// CheckCodec probes the recording file and checks if it needs re-encoding. +// Sets "needsReencode" to true if the video is not h264 or exceeds 3Mbit/s bitrate, +// or if audio is not AAC. +func CheckCodec(ctx context.Context, logger *slog.Logger, _ chan *protobuf.Notification, d map[string]any, _ *metrics.Broker) error { + recordingDir, ok := d["recordingDir"].(string) + if !ok { + return AbortingError(fmt.Errorf("no recordingDir in context")) + } + recording := path.Join(recordingDir, "playlist.m3u8") + + probe, err := ffmpeg.Probe(ctx, recording) + if err != nil { + return AbortingError(fmt.Errorf("ffprobe failed: %w", err)) + } + + needsReencode := false + for _, stream := range probe.Streams() { + if stream.CodecType == "video" { + if stream.CodecName != "h264" { + needsReencode = true + logger.Info("video codec requires re-encoding", "codec", stream.CodecName) + break + } + + if stream.BitRate > 3000000 { // 3 Mbit/s in bits/s + needsReencode = true + logger.Info("video bitrate exceeds 3Mbit/s", "bitrate", stream.BitRate) + break + } + } + + if stream.CodecType == "audio" { + if stream.CodecName != "aac" { + needsReencode = true + logger.Info("audio codec requires re-encoding", "codec", stream.CodecName) + break + } + } + } + + d["needsReencode"] = needsReencode + logger.Info("codec check completed", "needsReencode", needsReencode) + + return nil +} diff --git a/runner/pkg/actions/mkthumb.go b/runner/pkg/actions/mkthumb.go index 713e01df5..6454164b2 100644 --- a/runner/pkg/actions/mkthumb.go +++ b/runner/pkg/actions/mkthumb.go @@ -47,13 +47,12 @@ func MkThumb(_ context.Context, logger *slog.Logger, notify chan *protobuf.Notif return nil } -// createVideoThumbnail creates a thumbnail from the given video file - const ( thumbnailWidth = 720 // Width of the generated thumbnail in pixels jpegCompressionQuality = 90 // JPEG compression quality (0-100) ) +// createVideoThumbnail creates a thumbnail from the given video file func createVideoThumbnail(source string) ([]byte, error) { g, err := thumbgen.New(source, thumbnailWidth, 1, "", thumbgen.WithJpegCompression(jpegCompressionQuality)) if err != nil { diff --git a/runner/pkg/actions/mkvod.go b/runner/pkg/actions/mkvod.go index 4f69b110e..a2bf4d89f 100644 --- a/runner/pkg/actions/mkvod.go +++ b/runner/pkg/actions/mkvod.go @@ -31,9 +31,21 @@ func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notif if !ok { return AbortingError(fmt.Errorf("no stream version in context")) } - recordingDir, ok := d["recordingDir"].(string) - if !ok { - return AbortingError(fmt.Errorf("no recordingDir in context")) + var recording string + if rec, ok := d["recording"]; ok { + if recStr, ok := rec.(string); ok { + recording = recStr + } else { + return AbortingError(fmt.Errorf("recording value is not a string")) + } + } else if dir, ok := d["recordingDir"]; ok { + if dirStr, ok := dir.(string); ok { + recording = path.Join(dirStr, "playlist.m3u8") + } else { + return AbortingError(fmt.Errorf("recordingDir value is not a string")) + } + } else { + return AbortingError(fmt.Errorf("no recording or recordingDir in context")) } metrics.ConvertingProgresses.With(metrics.With().Stream(streamID).L()).Inc() @@ -45,7 +57,28 @@ func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notif return AbortingError(fmt.Errorf("create VOD directory: %w", err)) } - err = convertStream(ctx, logger, streamID, path.Join(recordingDir, "playlist.m3u8"), vodDir, "playlist.m3u8") + // Check if re-encoding is needed + var reencode bool + if needsReencode, ok := d["needsReencode"]; ok { + if reencodeVal, ok := needsReencode.(bool); ok { + reencode = reencodeVal + } else { + return AbortingError(fmt.Errorf("needsReencode is not a bool")) + } + } + + var videoCodec, audioCodec string + if reencode { + logger.Info("re-encoding required, transcoding video") + videoCodec = "libx264" + audioCodec = "aac" + } else { + logger.Info("no re-encoding needed, using copy codec") + videoCodec = "copy" + audioCodec = "copy" + } + + err = convertStream(ctx, logger, streamID, recording, vodDir, "playlist.m3u8", videoCodec, audioCodec) if err != nil { return AbortingError(fmt.Errorf("convert stream: %w", err)) } @@ -67,9 +100,18 @@ func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notif return nil } -func convertStream(ctx context.Context, logger *slog.Logger, streamID uint64, streamPath, vodDir string, playlistName string) error { +func convertStream(ctx context.Context, logger *slog.Logger, streamID uint64, streamPath, vodDir string, playlistName string, videoCodec string, audioCodec string) error { input := "-i " + streamPath - options := "-c copy -f hls -hls_time 20 -hls_playlist_type vod -hls_flags append_list -hls_segment_filename " + path.Join(vodDir, "%05d.ts") + " " + path.Join(vodDir, playlistName) + + // Build codec options based on parameters + codecOpts := fmt.Sprintf("-c:v %s -c:a %s", videoCodec, audioCodec) + + // Add bitrate limit if re-encoding video + if videoCodec != "copy" { + codecOpts += " -b:v 3M" + } + + options := codecOpts + " -f hls -hls_time 20 -hls_playlist_type vod -hls_flags append_list -hls_segment_filename " + path.Join(vodDir, "%05d.ts") + " " + path.Join(vodDir, playlistName) args := strings.Split(input, " ") args = append(args, strings.Split(options, " ")...) diff --git a/runner/pkg/ffmpeg/ffmpeg.go b/runner/pkg/ffmpeg/ffmpeg.go index d647b3a52..22fa46cfb 100644 --- a/runner/pkg/ffmpeg/ffmpeg.go +++ b/runner/pkg/ffmpeg/ffmpeg.go @@ -34,6 +34,30 @@ func (r *FFProbeResult) Container() string { return gjson.Get(r.raw, "format.format_name").String() } +// StreamInfo holds information about a single stream +type StreamInfo struct { + CodecType string + CodecName string + BitRate int64 +} + +// Streams returns all streams with their codec information +func (r *FFProbeResult) Streams() []StreamInfo { + nStreams := gjson.Get(r.raw, "streams.#").Int() + streams := make([]StreamInfo, 0, nStreams) + for i := 0; i < int(nStreams); i++ { + codecType := gjson.Get(r.raw, fmt.Sprintf("streams.%d.codec_type", i)).String() + codecName := gjson.Get(r.raw, fmt.Sprintf("streams.%d.codec_name", i)).String() + bitRate := gjson.Get(r.raw, fmt.Sprintf("streams.%d.bit_rate", i)).Int() + streams = append(streams, StreamInfo{ + CodecType: codecType, + CodecName: codecName, + BitRate: bitRate, + }) + } + return streams +} + // FFProbeResult holds the results of a ffprobe execution type FFProbeResult struct { raw string diff --git a/runner/protobuf/commons.pb.go b/runner/protobuf/commons.pb.go index 334b2fe15..81e984cfe 100644 --- a/runner/protobuf/commons.pb.go +++ b/runner/protobuf/commons.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.30.2 +// protoc-gen-go v1.36.11 +// protoc v6.33.2 // source: commons.proto package protobuf diff --git a/runner/protobuf/notifications.pb.go b/runner/protobuf/notifications.pb.go index 4cd1a4f97..a269f7f59 100644 --- a/runner/protobuf/notifications.pb.go +++ b/runner/protobuf/notifications.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.30.2 +// protoc-gen-go v1.36.11 +// protoc v6.33.2 // source: notifications.proto package protobuf diff --git a/runner/protobuf/runner.pb.go b/runner/protobuf/runner.pb.go index 563d89871..f14104890 100644 --- a/runner/protobuf/runner.pb.go +++ b/runner/protobuf/runner.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.30.2 +// protoc-gen-go v1.36.11 +// protoc v6.33.2 // source: runner.proto package protobuf @@ -334,6 +334,110 @@ func (*RegisterResponse) Descriptor() ([]byte, []int) { return file_runner_proto_rawDescGZIP(), []int{5} } +type HandleVODRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + StreamId *uint64 `protobuf:"varint,1,opt,name=stream_id,json=streamId" json:"stream_id,omitempty"` + Version *StreamVersion `protobuf:"varint,2,opt,name=version,enum=protobuf.StreamVersion" json:"version,omitempty"` + Filepath *string `protobuf:"bytes,3,opt,name=filepath" json:"filepath,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HandleVODRequest) Reset() { + *x = HandleVODRequest{} + mi := &file_runner_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HandleVODRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HandleVODRequest) ProtoMessage() {} + +func (x *HandleVODRequest) ProtoReflect() protoreflect.Message { + mi := &file_runner_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HandleVODRequest.ProtoReflect.Descriptor instead. +func (*HandleVODRequest) Descriptor() ([]byte, []int) { + return file_runner_proto_rawDescGZIP(), []int{6} +} + +func (x *HandleVODRequest) GetStreamId() uint64 { + if x != nil && x.StreamId != nil { + return *x.StreamId + } + return 0 +} + +func (x *HandleVODRequest) GetVersion() StreamVersion { + if x != nil && x.Version != nil { + return *x.Version + } + return StreamVersion_STREAM_VERSION_UNSPECIFIED +} + +func (x *HandleVODRequest) GetFilepath() string { + if x != nil && x.Filepath != nil { + return *x.Filepath + } + return "" +} + +type HandleVODResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + JobId *string `protobuf:"bytes,1,opt,name=job_id,json=jobId" json:"job_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HandleVODResponse) Reset() { + *x = HandleVODResponse{} + mi := &file_runner_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HandleVODResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HandleVODResponse) ProtoMessage() {} + +func (x *HandleVODResponse) ProtoReflect() protoreflect.Message { + mi := &file_runner_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HandleVODResponse.ProtoReflect.Descriptor instead. +func (*HandleVODResponse) Descriptor() ([]byte, []int) { + return file_runner_proto_rawDescGZIP(), []int{7} +} + +func (x *HandleVODResponse) GetJobId() string { + if x != nil && x.JobId != nil { + return *x.JobId + } + return "" +} + var File_runner_proto protoreflect.FileDescriptor const file_runner_proto_rawDesc = "" + @@ -356,10 +460,17 @@ const file_runner_proto_rawDesc = "" + "\bhostname\x18\x01 \x01(\tR\bhostname\x12\x12\n" + "\x04port\x18\x02 \x01(\x05R\x04port\x12\x18\n" + "\aversion\x18\x03 \x01(\tR\aversion\"\x12\n" + - "\x10RegisterResponse2\xa4\x01\n" + + "\x10RegisterResponse\"~\n" + + "\x10HandleVODRequest\x12\x1b\n" + + "\tstream_id\x18\x01 \x01(\x04R\bstreamId\x121\n" + + "\aversion\x18\x02 \x01(\x0e2\x17.protobuf.StreamVersionR\aversion\x12\x1a\n" + + "\bfilepath\x18\x03 \x01(\tR\bfilepath\"*\n" + + "\x11HandleVODResponse\x12\x15\n" + + "\x06job_id\x18\x01 \x01(\tR\x05jobId2\xec\x01\n" + "\rRunnerService\x12D\n" + "\rRequestStream\x12\x17.protobuf.StreamRequest\x1a\x18.protobuf.StreamResponse\"\x00\x12M\n" + - "\x10RequestStreamEnd\x12\x1a.protobuf.StreamEndRequest\x1a\x1b.protobuf.StreamEndResponse\"\x002\x9f\x01\n" + + "\x10RequestStreamEnd\x12\x1a.protobuf.StreamEndRequest\x1a\x1b.protobuf.StreamEndResponse\"\x00\x12F\n" + + "\tHandleVOD\x12\x1a.protobuf.HandleVODRequest\x1a\x1b.protobuf.HandleVODResponse\"\x002\x9f\x01\n" + "\x14RunnerManagerService\x12C\n" + "\bRegister\x12\x19.protobuf.RegisterRequest\x1a\x1a.protobuf.RegisterResponse\"\x00\x12B\n" + "\x06Notify\x12\x16.protobuf.Notification\x1a\x1e.protobuf.NotificationResponse\"\x00B\x11Z\x0frunner/protobufb\beditionsp\xe8\a" @@ -376,7 +487,7 @@ func file_runner_proto_rawDescGZIP() []byte { return file_runner_proto_rawDescData } -var file_runner_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_runner_proto_msgTypes = make([]protoimpl.MessageInfo, 8) var file_runner_proto_goTypes = []any{ (*StreamRequest)(nil), // 0: protobuf.StreamRequest (*StreamResponse)(nil), // 1: protobuf.StreamResponse @@ -384,27 +495,32 @@ var file_runner_proto_goTypes = []any{ (*StreamEndResponse)(nil), // 3: protobuf.StreamEndResponse (*RegisterRequest)(nil), // 4: protobuf.RegisterRequest (*RegisterResponse)(nil), // 5: protobuf.RegisterResponse - (StreamVersion)(0), // 6: protobuf.StreamVersion - (*timestamppb.Timestamp)(nil), // 7: google.protobuf.Timestamp - (*Notification)(nil), // 8: protobuf.Notification - (*NotificationResponse)(nil), // 9: protobuf.NotificationResponse + (*HandleVODRequest)(nil), // 6: protobuf.HandleVODRequest + (*HandleVODResponse)(nil), // 7: protobuf.HandleVODResponse + (StreamVersion)(0), // 8: protobuf.StreamVersion + (*timestamppb.Timestamp)(nil), // 9: google.protobuf.Timestamp + (*Notification)(nil), // 10: protobuf.Notification + (*NotificationResponse)(nil), // 11: protobuf.NotificationResponse } var file_runner_proto_depIdxs = []int32{ - 6, // 0: protobuf.StreamRequest.version:type_name -> protobuf.StreamVersion - 7, // 1: protobuf.StreamRequest.end:type_name -> google.protobuf.Timestamp - 0, // 2: protobuf.RunnerService.RequestStream:input_type -> protobuf.StreamRequest - 2, // 3: protobuf.RunnerService.RequestStreamEnd:input_type -> protobuf.StreamEndRequest - 4, // 4: protobuf.RunnerManagerService.Register:input_type -> protobuf.RegisterRequest - 8, // 5: protobuf.RunnerManagerService.Notify:input_type -> protobuf.Notification - 1, // 6: protobuf.RunnerService.RequestStream:output_type -> protobuf.StreamResponse - 3, // 7: protobuf.RunnerService.RequestStreamEnd:output_type -> protobuf.StreamEndResponse - 5, // 8: protobuf.RunnerManagerService.Register:output_type -> protobuf.RegisterResponse - 9, // 9: protobuf.RunnerManagerService.Notify:output_type -> protobuf.NotificationResponse - 6, // [6:10] is the sub-list for method output_type - 2, // [2:6] is the sub-list for method input_type - 2, // [2:2] is the sub-list for extension type_name - 2, // [2:2] is the sub-list for extension extendee - 0, // [0:2] is the sub-list for field type_name + 8, // 0: protobuf.StreamRequest.version:type_name -> protobuf.StreamVersion + 9, // 1: protobuf.StreamRequest.end:type_name -> google.protobuf.Timestamp + 8, // 2: protobuf.HandleVODRequest.version:type_name -> protobuf.StreamVersion + 0, // 3: protobuf.RunnerService.RequestStream:input_type -> protobuf.StreamRequest + 2, // 4: protobuf.RunnerService.RequestStreamEnd:input_type -> protobuf.StreamEndRequest + 6, // 5: protobuf.RunnerService.HandleVOD:input_type -> protobuf.HandleVODRequest + 4, // 6: protobuf.RunnerManagerService.Register:input_type -> protobuf.RegisterRequest + 10, // 7: protobuf.RunnerManagerService.Notify:input_type -> protobuf.Notification + 1, // 8: protobuf.RunnerService.RequestStream:output_type -> protobuf.StreamResponse + 3, // 9: protobuf.RunnerService.RequestStreamEnd:output_type -> protobuf.StreamEndResponse + 7, // 10: protobuf.RunnerService.HandleVOD:output_type -> protobuf.HandleVODResponse + 5, // 11: protobuf.RunnerManagerService.Register:output_type -> protobuf.RegisterResponse + 11, // 12: protobuf.RunnerManagerService.Notify:output_type -> protobuf.NotificationResponse + 8, // [8:13] is the sub-list for method output_type + 3, // [3:8] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name } func init() { file_runner_proto_init() } @@ -420,7 +536,7 @@ func file_runner_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_runner_proto_rawDesc), len(file_runner_proto_rawDesc)), NumEnums: 0, - NumMessages: 6, + NumMessages: 8, NumExtensions: 0, NumServices: 2, }, diff --git a/runner/protobuf/runner_grpc.pb.go b/runner/protobuf/runner_grpc.pb.go index 518ca9ecf..0f68fc441 100644 --- a/runner/protobuf/runner_grpc.pb.go +++ b/runner/protobuf/runner_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.5.1 -// - protoc v6.30.2 +// - protoc v6.33.2 // source: runner.proto package protobuf @@ -21,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9 const ( RunnerService_RequestStream_FullMethodName = "/protobuf.RunnerService/RequestStream" RunnerService_RequestStreamEnd_FullMethodName = "/protobuf.RunnerService/RequestStreamEnd" + RunnerService_HandleVOD_FullMethodName = "/protobuf.RunnerService/HandleVOD" ) // RunnerServiceClient is the client API for RunnerService service. @@ -30,6 +31,7 @@ type RunnerServiceClient interface { // Requests a stream from a lecture hall RequestStream(ctx context.Context, in *StreamRequest, opts ...grpc.CallOption) (*StreamResponse, error) RequestStreamEnd(ctx context.Context, in *StreamEndRequest, opts ...grpc.CallOption) (*StreamEndResponse, error) + HandleVOD(ctx context.Context, in *HandleVODRequest, opts ...grpc.CallOption) (*HandleVODResponse, error) } type runnerServiceClient struct { @@ -60,6 +62,16 @@ func (c *runnerServiceClient) RequestStreamEnd(ctx context.Context, in *StreamEn return out, nil } +func (c *runnerServiceClient) HandleVOD(ctx context.Context, in *HandleVODRequest, opts ...grpc.CallOption) (*HandleVODResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(HandleVODResponse) + err := c.cc.Invoke(ctx, RunnerService_HandleVOD_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // RunnerServiceServer is the server API for RunnerService service. // All implementations must embed UnimplementedRunnerServiceServer // for forward compatibility. @@ -67,6 +79,7 @@ type RunnerServiceServer interface { // Requests a stream from a lecture hall RequestStream(context.Context, *StreamRequest) (*StreamResponse, error) RequestStreamEnd(context.Context, *StreamEndRequest) (*StreamEndResponse, error) + HandleVOD(context.Context, *HandleVODRequest) (*HandleVODResponse, error) mustEmbedUnimplementedRunnerServiceServer() } @@ -83,6 +96,9 @@ func (UnimplementedRunnerServiceServer) RequestStream(context.Context, *StreamRe func (UnimplementedRunnerServiceServer) RequestStreamEnd(context.Context, *StreamEndRequest) (*StreamEndResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method RequestStreamEnd not implemented") } +func (UnimplementedRunnerServiceServer) HandleVOD(context.Context, *HandleVODRequest) (*HandleVODResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method HandleVOD not implemented") +} func (UnimplementedRunnerServiceServer) mustEmbedUnimplementedRunnerServiceServer() {} func (UnimplementedRunnerServiceServer) testEmbeddedByValue() {} @@ -140,6 +156,24 @@ func _RunnerService_RequestStreamEnd_Handler(srv interface{}, ctx context.Contex return interceptor(ctx, in, info, handler) } +func _RunnerService_HandleVOD_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(HandleVODRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(RunnerServiceServer).HandleVOD(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: RunnerService_HandleVOD_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(RunnerServiceServer).HandleVOD(ctx, req.(*HandleVODRequest)) + } + return interceptor(ctx, in, info, handler) +} + // RunnerService_ServiceDesc is the grpc.ServiceDesc for RunnerService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -155,6 +189,10 @@ var RunnerService_ServiceDesc = grpc.ServiceDesc{ MethodName: "RequestStreamEnd", Handler: _RunnerService_RequestStreamEnd_Handler, }, + { + MethodName: "HandleVOD", + Handler: _RunnerService_HandleVOD_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "runner.proto", diff --git a/runner/runner.proto b/runner/runner.proto index b928513be..32a3de2a7 100644 --- a/runner/runner.proto +++ b/runner/runner.proto @@ -12,6 +12,7 @@ service RunnerService { // Requests a stream from a lecture hall rpc RequestStream (StreamRequest) returns (StreamResponse) {} rpc RequestStreamEnd (StreamEndRequest) returns (StreamEndResponse) {} + rpc HandleVOD (HandleVODRequest) returns (HandleVODResponse) {} } message StreamRequest { @@ -52,3 +53,13 @@ message RegisterRequest { message RegisterResponse { } + +message HandleVODRequest { + uint64 stream_id = 1; + StreamVersion version = 2; + string filepath = 3; +} + +message HandleVODResponse { + string job_id = 1; +}