abc.txt

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.stereotype.Service;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.*;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.Collections;
import java.util.HashSet;

@Component
class MjpegFrameProcessor {
    private static final int BUFFER_SIZE = 8192;
    private final AtomicBoolean isRunning = new AtomicBoolean(true);
    private final Set<SseEmitter> emitters = Collections.synchronizedSet(new HashSet<>());
    private final ExecutorService executorService;
    private Future<?> processorFuture;

    public MjpegFrameProcessor() {
        // Use a single thread for frame processing to avoid overwhelming the system
        this.executorService = Executors.newSingleThreadExecutor();
    }

    public void startProcessing(String streamUrl) {
        processorFuture = executorService.submit(() -> {
            while (isRunning.get()) {
                try {
                    byte[] frame = captureFrame(streamUrl);
                    if (frame != null) {
                        broadcast(frame);
                    }
                } catch (IOException e) {
                    // Log error and possibly implement retry logic
                    e.printStackTrace();
                }
            }
        });
    }

    private byte[] captureFrame(String streamUrl) throws IOException {
        URL url = new URL(streamUrl);
        HttpURLConnection conn = (HttpURLConnection) url.openConnection();
        try (InputStream is = new BufferedInputStream(conn.getInputStream())) {
            // Read MJPEG frame boundary
            String boundary = findBoundary(is);
            if (boundary == null) return null;

            // Read frame content
            return readFrameContent(is, boundary);
        }
    }

    private String findBoundary(InputStream is) throws IOException {
        // Implementation to find MJPEG boundary marker
        // This is a simplified version - you'll need to implement proper boundary detection
        return "--myboundary";
    }

    private byte[] readFrameContent(InputStream is, String boundary) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        byte[] buffer = new byte[BUFFER_SIZE];
        int bytesRead;
        
        while ((bytesRead = is.read(buffer)) != -1) {
            baos.write(buffer, 0, bytesRead);
            // Check if we've reached the boundary
            if (new String(baos.toByteArray()).contains(boundary)) {
                break;
            }
        }
        
        return baos.toByteArray();
    }

    public SseEmitter addEmitter() {
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        emitters.add(emitter);
        
        emitter.onCompletion(() -> emitters.remove(emitter));
        emitter.onTimeout(() -> emitters.remove(emitter));
        emitter.onError(e -> emitters.remove(emitter));
        
        return emitter;
    }

    private void broadcast(byte[] frame) {
        Set<SseEmitter> deadEmitters = new HashSet<>();
        
        emitters.forEach(emitter -> {
            try {
                emitter.send(frame, MediaType.IMAGE_JPEG);
            } catch (IOException e) {
                deadEmitters.add(emitter);
            }
        });
        
        emitters.removeAll(deadEmitters);
    }

    @PreDestroy
    public void shutdown() {
        isRunning.set(false);
        if (processorFuture != null) {
            processorFuture.cancel(true);
        }
        executorService.shutdown();
    }
}

@RestController
@RequestMapping("/api/stream")
class StreamController {
    private final MjpegFrameProcessor frameProcessor;
    
    public StreamController(MjpegFrameProcessor frameProcessor) {
        this.frameProcessor = frameProcessor;
    }
    
    @GetMapping("/video")
    public SseEmitter streamVideo() {
        return frameProcessor.addEmitter();
    }
}

@Service
class StreamService {
    private final MjpegFrameProcessor frameProcessor;
    
    public StreamService(MjpegFrameProcessor frameProcessor) {
        this.frameProcessor = frameProcessor;
    }
    
    public void initializeStream(String streamUrl) {
        frameProcessor.startProcessing(streamUrl);
    }
}