forked from jsl/video_editing_poc
The storage port now has an opt-in S3-compatible implementation in src/main/java/org/example/videoclips/storage/S3ObjectStorageAdapter.java. It uses the AWS SDK presigner to generate multipart
upload part URLs and signed clip download URLs when video-clipping.storage=s3. The current in-memory adapter remains the default. I also added the supporting configuration fields in src/main/ java/org/example/videoclips/config/VideoClippingProperties.java and default values in src/main/resources/application.properties. The queue port now has a durable database-backed implementation in src/main/java/org/example/videoclips/queue/DatabaseBackedClipJobQueueAdapter.java. It persists queue messages in a new queue_messages table from src/main/resources/db/migration/V2__queue_messages.sql, polls them on a schedule, and dispatches them through the existing processor. I updated src/main/java/org/ example/videoclips/processing/ClipProcessor.java so processing failures propagate back to the queue adapter instead of being swallowed. I also enabled scheduling in src/main/java/org/example/videoclips/VideoClippingApplication.java, added the AWS SDK dependency in pom.xml, and updated src/main/resources/application-jpa.properties so the JPA profile uses the database queue by default.
This commit is contained in:
JSLMPR
parent
517a4e1146
commit
67c7b65639
17
pom.xml
17
pom.xml
|
|
@ -19,8 +19,21 @@
|
|||
<properties>
|
||||
<java.version>21</java.version>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<aws.sdk.version>2.28.29</aws.sdk.version>
|
||||
</properties>
|
||||
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>software.amazon.awssdk</groupId>
|
||||
<artifactId>bom</artifactId>
|
||||
<version>${aws.sdk.version}</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
|
|
@ -52,6 +65,10 @@
|
|||
<artifactId>h2</artifactId>
|
||||
<scope>runtime</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>software.amazon.awssdk</groupId>
|
||||
<artifactId>s3</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
|
|
|
|||
|
|
@ -4,9 +4,11 @@ import org.springframework.boot.SpringApplication;
|
|||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
|
||||
import org.springframework.scheduling.annotation.EnableAsync;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
|
||||
@SpringBootApplication
|
||||
@EnableAsync
|
||||
@EnableScheduling
|
||||
@ConfigurationPropertiesScan
|
||||
public class VideoClippingApplication {
|
||||
|
||||
|
|
|
|||
|
|
@ -24,6 +24,14 @@ public class VideoClippingProperties {
|
|||
|
||||
private List<String> allowedContentTypes = List.of("video/mp4", "video/quicktime", "video/webm");
|
||||
|
||||
private String storage = "memory";
|
||||
|
||||
private String queue = "memory";
|
||||
|
||||
private final S3 s3 = new S3();
|
||||
|
||||
private final DatabaseQueue databaseQueue = new DatabaseQueue();
|
||||
|
||||
public long getMaxFileSizeBytes() {
|
||||
return maxFileSizeBytes;
|
||||
}
|
||||
|
|
@ -63,4 +71,91 @@ public class VideoClippingProperties {
|
|||
public void setAllowedContentTypes(List<String> allowedContentTypes) {
|
||||
this.allowedContentTypes = allowedContentTypes;
|
||||
}
|
||||
|
||||
public String getStorage() {
|
||||
return storage;
|
||||
}
|
||||
|
||||
public void setStorage(String storage) {
|
||||
this.storage = storage;
|
||||
}
|
||||
|
||||
public String getQueue() {
|
||||
return queue;
|
||||
}
|
||||
|
||||
public void setQueue(String queue) {
|
||||
this.queue = queue;
|
||||
}
|
||||
|
||||
public S3 getS3() {
|
||||
return s3;
|
||||
}
|
||||
|
||||
public DatabaseQueue getDatabaseQueue() {
|
||||
return databaseQueue;
|
||||
}
|
||||
|
||||
public static class S3 {
|
||||
private String region = "us-east-1";
|
||||
private String bucket = "video-clipping-dev";
|
||||
private String endpoint;
|
||||
private boolean pathStyle = true;
|
||||
|
||||
public String getRegion() {
|
||||
return region;
|
||||
}
|
||||
|
||||
public void setRegion(String region) {
|
||||
this.region = region;
|
||||
}
|
||||
|
||||
public String getBucket() {
|
||||
return bucket;
|
||||
}
|
||||
|
||||
public void setBucket(String bucket) {
|
||||
this.bucket = bucket;
|
||||
}
|
||||
|
||||
public String getEndpoint() {
|
||||
return endpoint;
|
||||
}
|
||||
|
||||
public void setEndpoint(String endpoint) {
|
||||
this.endpoint = endpoint;
|
||||
}
|
||||
|
||||
public boolean isPathStyle() {
|
||||
return pathStyle;
|
||||
}
|
||||
|
||||
public void setPathStyle(boolean pathStyle) {
|
||||
this.pathStyle = pathStyle;
|
||||
}
|
||||
}
|
||||
|
||||
public static class DatabaseQueue {
|
||||
@Min(100)
|
||||
private long pollIntervalMs = 1000;
|
||||
|
||||
@Min(1)
|
||||
private int batchSize = 5;
|
||||
|
||||
public long getPollIntervalMs() {
|
||||
return pollIntervalMs;
|
||||
}
|
||||
|
||||
public void setPollIntervalMs(long pollIntervalMs) {
|
||||
this.pollIntervalMs = pollIntervalMs;
|
||||
}
|
||||
|
||||
public int getBatchSize() {
|
||||
return batchSize;
|
||||
}
|
||||
|
||||
public void setBatchSize(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,6 +25,10 @@ public class ClipProcessor {
|
|||
|
||||
@Async
|
||||
public void process(String jobId) {
|
||||
processNow(jobId);
|
||||
}
|
||||
|
||||
public void processNow(String jobId) {
|
||||
try {
|
||||
ClipJob job = requireJob(jobId);
|
||||
repository.saveJob(job.withStatus(ClipJobStatus.RUNNING).withStartedAt(Instant.now()));
|
||||
|
|
@ -65,6 +69,7 @@ public class ClipProcessor {
|
|||
}
|
||||
recordEvent(jobId, "FAILED", ex.getMessage());
|
||||
}
|
||||
throw new IllegalStateException("Clip processing failed for job " + jobId, ex);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,71 @@
|
|||
package org.example.videoclips.queue;
|
||||
|
||||
import org.example.videoclips.config.VideoClippingProperties;
|
||||
import org.example.videoclips.processing.ClipProcessor;
|
||||
import org.example.videoclips.queue.entity.QueueMessageEntity;
|
||||
import org.example.videoclips.queue.spring.QueueMessageJpaRepository;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
@Component
|
||||
@ConditionalOnProperty(name = "video-clipping.queue", havingValue = "db")
|
||||
public class DatabaseBackedClipJobQueueAdapter implements ClipJobQueuePort {
|
||||
|
||||
private final QueueMessageJpaRepository queueMessageJpaRepository;
|
||||
private final ClipProcessor clipProcessor;
|
||||
private final VideoClippingProperties properties;
|
||||
|
||||
public DatabaseBackedClipJobQueueAdapter(
|
||||
QueueMessageJpaRepository queueMessageJpaRepository,
|
||||
ClipProcessor clipProcessor,
|
||||
VideoClippingProperties properties
|
||||
) {
|
||||
this.queueMessageJpaRepository = queueMessageJpaRepository;
|
||||
this.clipProcessor = clipProcessor;
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void publish(String jobId) {
|
||||
QueueMessageEntity entity = new QueueMessageEntity();
|
||||
entity.setId("msg_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12));
|
||||
entity.setJobId(jobId);
|
||||
entity.setStatus("PENDING");
|
||||
entity.setAttemptCount(0);
|
||||
entity.setAvailableAt(Instant.now());
|
||||
queueMessageJpaRepository.save(entity);
|
||||
}
|
||||
|
||||
@Scheduled(fixedDelayString = "${video-clipping.database-queue.poll-interval-ms:1000}")
|
||||
@Transactional
|
||||
public void poll() {
|
||||
int remaining = properties.getDatabaseQueue().getBatchSize();
|
||||
List<QueueMessageEntity> messages = queueMessageJpaRepository
|
||||
.findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc("PENDING", Instant.now());
|
||||
for (QueueMessageEntity message : messages) {
|
||||
if (remaining-- <= 0) {
|
||||
break;
|
||||
}
|
||||
message.setStatus("PROCESSING");
|
||||
message.setClaimedAt(Instant.now());
|
||||
message.setAttemptCount(message.getAttemptCount() + 1);
|
||||
queueMessageJpaRepository.save(message);
|
||||
try {
|
||||
clipProcessor.processNow(message.getJobId());
|
||||
message.setStatus("COMPLETED");
|
||||
message.setLastError(null);
|
||||
} catch (Exception ex) {
|
||||
message.setStatus("FAILED");
|
||||
message.setLastError(ex.getMessage());
|
||||
}
|
||||
queueMessageJpaRepository.save(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2,9 +2,11 @@ package org.example.videoclips.queue;
|
|||
|
||||
import org.example.videoclips.processing.ClipProcessor;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
@ConditionalOnProperty(name = "video-clipping.queue", havingValue = "memory", matchIfMissing = true)
|
||||
public class InMemoryClipJobQueueAdapter implements ClipJobQueuePort {
|
||||
|
||||
private final ClipProcessor clipProcessor;
|
||||
|
|
@ -16,6 +18,6 @@ public class InMemoryClipJobQueueAdapter implements ClipJobQueuePort {
|
|||
@Override
|
||||
@Async
|
||||
public void publish(String jobId) {
|
||||
clipProcessor.process(jobId);
|
||||
clipProcessor.processNow(jobId);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,89 @@
|
|||
package org.example.videoclips.queue.entity;
|
||||
|
||||
import jakarta.persistence.Column;
|
||||
import jakarta.persistence.Entity;
|
||||
import jakarta.persistence.Id;
|
||||
import jakarta.persistence.Table;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
@Entity
|
||||
@Table(name = "queue_messages")
|
||||
public class QueueMessageEntity {
|
||||
|
||||
@Id
|
||||
private String id;
|
||||
|
||||
@Column(nullable = false, length = 64)
|
||||
private String jobId;
|
||||
|
||||
@Column(nullable = false, length = 32)
|
||||
private String status;
|
||||
|
||||
@Column(nullable = false)
|
||||
private int attemptCount;
|
||||
|
||||
@Column(nullable = false)
|
||||
private Instant availableAt;
|
||||
|
||||
private Instant claimedAt;
|
||||
|
||||
@Column(length = 1000)
|
||||
private String lastError;
|
||||
|
||||
public String getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(String id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getJobId() {
|
||||
return jobId;
|
||||
}
|
||||
|
||||
public void setJobId(String jobId) {
|
||||
this.jobId = jobId;
|
||||
}
|
||||
|
||||
public String getStatus() {
|
||||
return status;
|
||||
}
|
||||
|
||||
public void setStatus(String status) {
|
||||
this.status = status;
|
||||
}
|
||||
|
||||
public int getAttemptCount() {
|
||||
return attemptCount;
|
||||
}
|
||||
|
||||
public void setAttemptCount(int attemptCount) {
|
||||
this.attemptCount = attemptCount;
|
||||
}
|
||||
|
||||
public Instant getAvailableAt() {
|
||||
return availableAt;
|
||||
}
|
||||
|
||||
public void setAvailableAt(Instant availableAt) {
|
||||
this.availableAt = availableAt;
|
||||
}
|
||||
|
||||
public Instant getClaimedAt() {
|
||||
return claimedAt;
|
||||
}
|
||||
|
||||
public void setClaimedAt(Instant claimedAt) {
|
||||
this.claimedAt = claimedAt;
|
||||
}
|
||||
|
||||
public String getLastError() {
|
||||
return lastError;
|
||||
}
|
||||
|
||||
public void setLastError(String lastError) {
|
||||
this.lastError = lastError;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
package org.example.videoclips.queue.spring;
|
||||
|
||||
import org.example.videoclips.queue.entity.QueueMessageEntity;
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public interface QueueMessageJpaRepository extends JpaRepository<QueueMessageEntity, String> {
|
||||
|
||||
List<QueueMessageEntity> findTop10ByStatusAndAvailableAtLessThanEqualOrderByAvailableAtAsc(String status, java.time.Instant availableAt);
|
||||
}
|
||||
|
|
@ -1,11 +1,13 @@
|
|||
package org.example.videoclips.storage;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
|
||||
@Component
|
||||
@ConditionalOnProperty(name = "video-clipping.storage", havingValue = "memory", matchIfMissing = true)
|
||||
public class InMemoryObjectStorageAdapter implements ObjectStoragePort {
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -0,0 +1,83 @@
|
|||
package org.example.videoclips.storage;
|
||||
|
||||
import org.example.videoclips.config.VideoClippingProperties;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
|
||||
import software.amazon.awssdk.regions.Region;
|
||||
import software.amazon.awssdk.services.s3.S3Configuration;
|
||||
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
|
||||
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
|
||||
import software.amazon.awssdk.services.s3.presigner.S3Presigner;
|
||||
import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;
|
||||
import software.amazon.awssdk.services.s3.presigner.model.PresignedGetObjectRequest;
|
||||
import software.amazon.awssdk.services.s3.presigner.model.PresignedUploadPartRequest;
|
||||
import software.amazon.awssdk.services.s3.presigner.model.UploadPartPresignRequest;
|
||||
|
||||
import java.net.URI;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.stream.IntStream;
|
||||
|
||||
@Component
|
||||
@ConditionalOnProperty(name = "video-clipping.storage", havingValue = "s3")
|
||||
public class S3ObjectStorageAdapter implements ObjectStoragePort {
|
||||
|
||||
private final VideoClippingProperties properties;
|
||||
private final S3Presigner s3Presigner;
|
||||
|
||||
public S3ObjectStorageAdapter(VideoClippingProperties properties) {
|
||||
this.properties = properties;
|
||||
Region region = Region.of(properties.getS3().getRegion());
|
||||
S3Presigner.Builder presignerBuilder = S3Presigner.builder()
|
||||
.region(region)
|
||||
.serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(properties.getS3().isPathStyle()).build())
|
||||
.credentialsProvider(DefaultCredentialsProvider.create());
|
||||
if (properties.getS3().getEndpoint() != null && !properties.getS3().getEndpoint().isBlank()) {
|
||||
URI endpoint = URI.create(properties.getS3().getEndpoint());
|
||||
presignerBuilder = presignerBuilder.endpointOverride(endpoint);
|
||||
}
|
||||
this.s3Presigner = presignerBuilder.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public UploadSessionDescriptor createMultipartUpload(String uploadId, int partCount, long partSizeBytes, Instant expiresAt) {
|
||||
String objectKey = "uploads/" + uploadId + "/source";
|
||||
Duration signatureDuration = durationUntil(expiresAt);
|
||||
List<UploadPartDescriptor> parts = IntStream.rangeClosed(1, partCount)
|
||||
.mapToObj(partNumber -> {
|
||||
UploadPartRequest uploadPartRequest = UploadPartRequest.builder()
|
||||
.bucket(properties.getS3().getBucket())
|
||||
.key(objectKey)
|
||||
.uploadId(uploadId)
|
||||
.partNumber(partNumber)
|
||||
.build();
|
||||
PresignedUploadPartRequest presignedRequest = s3Presigner.presignUploadPart(UploadPartPresignRequest.builder()
|
||||
.signatureDuration(signatureDuration)
|
||||
.uploadPartRequest(uploadPartRequest)
|
||||
.build());
|
||||
return new UploadPartDescriptor(partNumber, presignedRequest.url().toString());
|
||||
})
|
||||
.toList();
|
||||
return new UploadSessionDescriptor(uploadId, "multipart", partSizeBytes, expiresAt, parts);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String createClipDownloadUrl(String clipId, Instant expiresAt) {
|
||||
String objectKey = "clips/" + clipId + ".mp4";
|
||||
PresignedGetObjectRequest presignedRequest = s3Presigner.presignGetObject(GetObjectPresignRequest.builder()
|
||||
.signatureDuration(durationUntil(expiresAt))
|
||||
.getObjectRequest(GetObjectRequest.builder()
|
||||
.bucket(properties.getS3().getBucket())
|
||||
.key(objectKey)
|
||||
.build())
|
||||
.build());
|
||||
return presignedRequest.url().toString();
|
||||
}
|
||||
|
||||
private Duration durationUntil(Instant expiresAt) {
|
||||
Duration duration = Duration.between(Instant.now(), expiresAt);
|
||||
return duration.isNegative() ? Duration.ofMinutes(1) : duration;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,5 @@
|
|||
video-clipping.repository=jpa
|
||||
video-clipping.queue=db
|
||||
|
||||
spring.datasource.url=jdbc:h2:mem:videoclips;MODE=PostgreSQL;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
|
||||
spring.datasource.driver-class-name=org.h2.Driver
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ spring.mvc.problemdetails.enabled=true
|
|||
management.endpoints.web.exposure.include=health,info,metrics
|
||||
spring.jpa.open-in-view=false
|
||||
video-clipping.repository=memory
|
||||
video-clipping.storage=memory
|
||||
video-clipping.queue=memory
|
||||
video-clipping.max-file-size-bytes=53687091200
|
||||
video-clipping.multipart-part-size-bytes=104857600
|
||||
video-clipping.upload-url-ttl-minutes=60
|
||||
|
|
@ -10,3 +12,8 @@ video-clipping.download-url-ttl-minutes=15
|
|||
video-clipping.allowed-content-types[0]=video/mp4
|
||||
video-clipping.allowed-content-types[1]=video/quicktime
|
||||
video-clipping.allowed-content-types[2]=video/webm
|
||||
video-clipping.s3.region=us-east-1
|
||||
video-clipping.s3.bucket=video-clipping-dev
|
||||
video-clipping.s3.path-style=true
|
||||
video-clipping.database-queue.poll-interval-ms=1000
|
||||
video-clipping.database-queue.batch-size=5
|
||||
|
|
|
|||
|
|
@ -0,0 +1,12 @@
|
|||
create table if not exists queue_messages (
|
||||
id varchar(64) primary key,
|
||||
job_id varchar(64) not null,
|
||||
status varchar(32) not null,
|
||||
attempt_count integer not null,
|
||||
available_at timestamp with time zone not null,
|
||||
claimed_at timestamp with time zone,
|
||||
last_error varchar(1000)
|
||||
);
|
||||
|
||||
create index if not exists idx_queue_messages_status_available_at
|
||||
on queue_messages(status, available_at);
|
||||
Loading…
Reference in New Issue