feat(streaming): add backend streaming support with nginx-rtmp #1

Merged
dylan merged 31 commits from feat/rtmp-streaming into master 2026-06-20 01:33:56 +01:00
12 changed files with 170 additions and 12 deletions
Showing only changes of commit 1dbc40d8e6 - Show all commits

View File

@@ -1,11 +1,9 @@
FROM tiangolo/nginx-rtmp FROM tiangolo/nginx-rtmp
RUN apt-get update && apt-get install -y gettext-base && rm -rf /var/lib/apt/lists/* RUN apt-get update && apt-get install -y gettext-base && rm -rf /var/lib/apt/lists/*
COPY default.conf.template /etc/nginx/templates/default.conf.template
COPY docker-entrypoint.sh /docker-entrypoint.sh
COPY nginx.conf /etc/nginx/nginx.conf.template RUN chmod +x /docker-entrypoint.sh
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
EXPOSE 1935 ENTRYPOINT ["/docker-entrypoint.sh"]
ENTRYPOINT ["/entrypoint.sh"]

View File

@@ -0,0 +1,9 @@
#!/bin/sh
# must only run on certain env files, otherwise it replaces too much
envsubst '${WEB_SERVER_URL}' \
< /etc/nginx/templates/default.conf.template \
> /etc/nginx/nginx.conf
# daemon off required otherwise the container would exit immediately after nginx forks to background.
exec nginx -g "daemon off;"

View File

@@ -35,6 +35,7 @@ public class SecurityConfig {
.requestMatchers("/api/v1/upload", "/api/v1/download/**").permitAll() .requestMatchers("/api/v1/upload", "/api/v1/download/**").permitAll()
.requestMatchers("/api/v1/edit/**", "/api/v1/process/**", "/api/v1/progress/**", "/api/v1/convert/**").permitAll() .requestMatchers("/api/v1/edit/**", "/api/v1/process/**", "/api/v1/progress/**", "/api/v1/convert/**").permitAll()
.requestMatchers("/api/v1/metadata/**").permitAll() .requestMatchers("/api/v1/metadata/**").permitAll()
.requestMatchers("/api/v1/stream/start", "/api/v1/stream/stop", "/api/v1/stream/heartbeat").permitAll()
.anyRequest().authenticated() .anyRequest().authenticated()
) )
.sessionManagement(session -> session.sessionCreationPolicy(SessionCreationPolicy.STATELESS) .sessionManagement(session -> session.sessionCreationPolicy(SessionCreationPolicy.STATELESS)

View File

@@ -1,9 +1,11 @@
package com.ddf.vodsystem.controllers; package com.ddf.vodsystem.controllers;
import com.ddf.vodsystem.dto.APIResponse; import com.ddf.vodsystem.dto.APIResponse;
import com.ddf.vodsystem.exceptions.AlreadyStreaming;
import com.ddf.vodsystem.exceptions.FFMPEGException; import com.ddf.vodsystem.exceptions.FFMPEGException;
import com.ddf.vodsystem.exceptions.JobNotFinished; import com.ddf.vodsystem.exceptions.JobNotFinished;
import com.ddf.vodsystem.exceptions.JobNotFound; import com.ddf.vodsystem.exceptions.JobNotFound;
import com.ddf.vodsystem.exceptions.KeyNotFound;
import com.ddf.vodsystem.exceptions.NotAuthenticated; import com.ddf.vodsystem.exceptions.NotAuthenticated;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity; import org.springframework.http.ResponseEntity;
@@ -76,4 +78,18 @@ public class GlobalExceptionHandler {
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null); APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.UNAUTHORIZED).body(response); return ResponseEntity.status(HttpStatus.UNAUTHORIZED).body(response);
} }
@ExceptionHandler(KeyNotFound.class)
public ResponseEntity<APIResponse<Void>> handleKeyNotFound(KeyNotFound ex) {
logger.error("KeyNotFound: {}", ex.getMessage());
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.NOT_FOUND).body(response);
}
@ExceptionHandler(AlreadyStreaming.class)
public ResponseEntity<APIResponse<Void>> handleAlreadyStreaming(AlreadyStreaming ex) {
logger.error("AlreadyStreaming: {}", ex.getMessage());
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.CONFLICT).body(response);
}
} }

View File

@@ -0,0 +1,52 @@
package com.ddf.vodsystem.controllers;
import com.ddf.vodsystem.dto.APIResponse;
import com.ddf.vodsystem.entities.Stream;
import com.ddf.vodsystem.services.StreamService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.List;
@RestController
@RequestMapping("/api/v1/stream")
public class StreamController {
private static final Logger logger = LoggerFactory.getLogger(StreamController.class);
private final StreamService streamService;
public StreamController(StreamService streamService) {
this.streamService = streamService;
}
// Called by nginx on_publish
@PostMapping("/start")
public ResponseEntity<APIResponse<Stream>> startStream(@RequestParam("name") String streamKey) {
logger.info("Stream start callback for key: {}", streamKey);
Stream stream = streamService.startStream(streamKey);
return ResponseEntity.ok(new APIResponse<>("success", "Stream started", stream));
}
// Called by nginx on_publish_done
@PostMapping("/stop")
public ResponseEntity<APIResponse<Void>> stopStream(@RequestParam("name") String streamKey) {
logger.info("Stream stop callback for key: {}", streamKey);
streamService.endStream(streamKey);
return ResponseEntity.ok(new APIResponse<>("success", "Stream ended", null));
}
// Called by nginx on_update
@PostMapping("/heartbeat")
public ResponseEntity<APIResponse<Void>> heartbeat(@RequestParam("name") String streamKey) {
streamService.heartbeatStream(streamKey);
return ResponseEntity.ok(new APIResponse<>("success", "Heartbeat recorded", null));
}
@GetMapping("/history/{userId}")
public ResponseEntity<APIResponse<List<Stream>>> getStreamHistory(@PathVariable Long userId) {
List<Stream> history = streamService.getStreamHistory(userId);
return ResponseEntity.ok(new APIResponse<>("success", "Stream history retrieved", history));
}
}

View File

@@ -0,0 +1,7 @@
package com.ddf.vodsystem.exceptions;
public class AlreadyStreaming extends RuntimeException {
public AlreadyStreaming(String message) {
super(message);
}
}

View File

@@ -0,0 +1,7 @@
package com.ddf.vodsystem.exceptions;
public class KeyNotFound extends RuntimeException {
public KeyNotFound(String message) {
super(message);
}
}

View File

@@ -1,9 +1,15 @@
package com.ddf.vodsystem.repositories; package com.ddf.vodsystem.repositories;
import com.ddf.vodsystem.entities.Stream; import com.ddf.vodsystem.entities.Stream;
import com.ddf.vodsystem.entities.User;
import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
import java.util.List;
import java.util.Optional;
@Repository @Repository
public interface StreamRepository extends JpaRepository<Stream, Long> { public interface StreamRepository extends JpaRepository<Stream, Long> {
Optional<Stream> findByUserAndEndDateIsNull(User user);
List<Stream> findByUser(User user);
} }

View File

@@ -1,24 +1,77 @@
package com.ddf.vodsystem.services; package com.ddf.vodsystem.services;
import com.ddf.vodsystem.entities.Stream;
import com.ddf.vodsystem.entities.User; import com.ddf.vodsystem.entities.User;
import com.ddf.vodsystem.exceptions.AlreadyStreaming;
import com.ddf.vodsystem.exceptions.KeyNotFound;
import com.ddf.vodsystem.repositories.StreamRepository; import com.ddf.vodsystem.repositories.StreamRepository;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Optional;
@Service
public class StreamService { public class StreamService {
private StreamRepository streamRepository; private final StreamRepository streamRepository;
private final UserService userService;
public StreamService(StreamRepository streamRepository) { public StreamService(StreamRepository streamRepository, UserService userService) {
this.streamRepository = streamRepository; this.streamRepository = streamRepository;
this.userService = userService;
} }
public void startStream(User user) { public Stream startStream(String streamKey) {
User user = resolveUser(streamKey);
streamRepository.findByUserAndEndDateIsNull(user).ifPresent(existing -> {
throw new AlreadyStreaming("User " + user.getUsername() + " is already streaming.");
});
LocalDateTime now = LocalDateTime.now();
Stream stream = new Stream();
stream.setUser(user);
stream.setStartDate(now);
stream.setLastSeen(now);
return streamRepository.saveAndFlush(stream);
} }
public void endStream(User user) { public void endStream(String streamKey) {
User user = resolveUser(streamKey);
Stream stream = streamRepository.findByUserAndEndDateIsNull(user)
.orElseThrow(() -> new IllegalStateException(
"No active stream found for user " + user.getUsername()));
stream.setEndDate(LocalDateTime.now());
streamRepository.saveAndFlush(stream);
} }
public void heartbeatStream(User user) { public void heartbeatStream(String streamKey) {
User user = resolveUser(streamKey);
Stream stream = streamRepository.findByUserAndEndDateIsNull(user)
.orElseThrow(() -> new IllegalStateException(
"No active stream found for user " + user.getUsername()));
stream.setLastSeen(LocalDateTime.now());
streamRepository.saveAndFlush(stream);
}
public Optional<Stream> getActiveStream(String streamKey) {
User user = resolveUser(streamKey);
return streamRepository.findByUserAndEndDateIsNull(user);
}
public List<Stream> getStreamHistory(Long userId) {
User user = userService.getUserById(userId)
.orElseThrow(() -> new IllegalArgumentException("User not found: " + userId));
return streamRepository.findByUser(user);
}
private User resolveUser(String streamKey) {
return userService.getUserByStreamKey(streamKey)
.orElseThrow(() -> new KeyNotFound("Stream key not found: " + streamKey));
} }
} }

View File

@@ -16,8 +16,10 @@ import org.springframework.stereotype.Service;
import java.io.IOException; import java.io.IOException;
import java.security.GeneralSecurityException; import java.security.GeneralSecurityException;
import java.security.SecureRandom;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.Collections; import java.util.Collections;
import java.util.HexFormat;
import java.util.Optional; import java.util.Optional;
@Service @Service
@@ -77,6 +79,7 @@ public class UserService {
if (existingUser.isEmpty()) { if (existingUser.isEmpty()) {
user.setRole(0); user.setRole(0);
user.setCreatedAt(LocalDateTime.now()); user.setCreatedAt(LocalDateTime.now());
user.setStreamKey(generateStreamKey());
return userRepository.saveAndFlush(user); return userRepository.saveAndFlush(user);
} }
@@ -111,4 +114,10 @@ public class UserService {
throw new NotAuthenticated("Invalid ID token: " + e.getMessage()); throw new NotAuthenticated("Invalid ID token: " + e.getMessage());
} }
} }
private String generateStreamKey() {
byte[] bytes = new byte[24];
new SecureRandom().nextBytes(bytes);
return HexFormat.of().formatHex(bytes);
}
} }

View File

@@ -11,7 +11,7 @@ CREATE TABLE IF NOT EXISTS users (
profile_picture_url VARCHAR(255), profile_picture_url VARCHAR(255),
role INTEGER DEFAULT 0, role INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
stream_key VARCHAR(64) stream_key VARCHAR(64) NOT NULL
); );
CREATE TABLE IF NOT EXISTS streams ( CREATE TABLE IF NOT EXISTS streams (