Implement video rotator

Now when the client publishes a stream that contains a rotate prefix
in the name the video is rotated using FFmpeg and re-published as
without the prefix.
Valid prefixes are: "rotate_left/" and "rotate_right/".

bbb-video/src/main/java/org/bigbluebutton/app/video/VideoApplication.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.bigbluebutton.app.video.converter.H263Converter;
+import org.bigbluebutton.app.video.converter.VideoRotator;
 import org.red5.logging.Red5LoggerFactory;
 import org.red5.server.adapter.MultiThreadedApplicationAdapter;
 import org.red5.server.api.IConnection;
@@ -49,6 +50,8 @@ public class VideoApplication extends MultiThreadedApplicationAdapter {
 
 	private final Map<String, H263Converter> h263Converters = new HashMap<String, H263Converter>();
 
+	private final Map<String, VideoRotator> videoRotators = new HashMap<String, VideoRotator>();
+
     @Override
 	public boolean appStart(IScope app) {
 	    super.appStart(app);
@@ -179,14 +182,21 @@ public void streamPublishStart(IBroadcastStream stream) {
     public void streamBroadcastStart(IBroadcastStream stream) {
     	IConnection conn = Red5.getConnectionLocal();  
     	super.streamBroadcastStart(stream);
-    	log.info("streamBroadcastStart " + stream.getPublishedName() + " " + System.currentTimeMillis() + " " + conn.getScope().getName());
+    	String streamName = stream.getPublishedName();
+    	log.info("streamBroadcastStart " + streamName + " " + System.currentTimeMillis() + " " + conn.getScope().getName());
 
-        if (recordVideoStream && stream.getPublishedName().contains("/") == false) {
+        if (streamName.contains("/")) {
+            if(VideoRotator.getDirection(streamName) != null) {
+                VideoRotator rotator = new VideoRotator(streamName);
+                videoRotators.put(streamName, rotator);
+            }
+        }
+        else if (recordVideoStream) {
 	    	recordStream(stream);
 	    	VideoStreamListener listener = new VideoStreamListener(); 
 	        listener.setEventRecordingService(recordingService);
 	        stream.addStreamListener(listener); 
-	        streamListeners.put(conn.getScope().getName() + "-" + stream.getPublishedName(), listener);
+	        streamListeners.put(conn.getScope().getName() + "-" + streamName, listener);
         }
     }
 
@@ -271,9 +281,15 @@ public void streamBroadcastClose(IBroadcastStream stream) {
         recordingService.record(scopeName, event);    		
       }
 
-		if(h263Converters.containsKey(stream.getName())) {
+		String streamName = stream.getName();
+		if(h263Converters.containsKey(streamName)) {
 			// Stop converter
-			h263Converters.remove(stream.getName()).stopConverter();
+			h263Converters.remove(streamName).stopConverter();
+		}
+
+		if(videoRotators.containsKey(streamName)) {
+			// Stop rotator
+			videoRotators.remove(streamName).stop();
 		}
     }
 

bbb-video/src/main/java/org/bigbluebutton/app/video/converter/VideoRotator.java

@@ -0,0 +1,110 @@
+package org.bigbluebutton.app.video.converter;
+
+import org.bigbluebutton.app.video.ffmpeg.FFmpegCommand;
+import org.bigbluebutton.app.video.ffmpeg.ProcessMonitor;
+import org.red5.logging.Red5LoggerFactory;
+import org.red5.server.api.IConnection;
+import org.red5.server.api.Red5;
+import org.slf4j.Logger;
+
+/**
+ * Represents a stream rotator. This class is responsible
+ * for choosing the rotate direction based on the stream name
+ * and starting FFmpeg to rotate and re-publish the stream.
+ */
+public class VideoRotator {
+
+	private static Logger log = Red5LoggerFactory.getLogger(VideoRotator.class, "video");
+
+	public static final String ROTATE_LEFT = "rotate_left";
+	public static final String ROTATE_RIGHT = "rotate_right";
+
+	private String streamName;
+	private FFmpegCommand.ROTATE direction;
+
+	private FFmpegCommand ffmpeg;
+	private ProcessMonitor processMonitor;
+
+	/**
+	 * Create a new video rotator for the specified stream.
+	 * The streamName should be of the form: 
+	 * rotate_[left|right]/streamName
+	 * The rotated stream will be published as streamName.
+	 * 
+	 * @param origin Name of the stream that will be rotated
+	 */
+	public VideoRotator(String origin) {
+		this.streamName = getStreamName(origin);
+		this.direction = getDirection(origin);
+
+		IConnection conn = Red5.getConnectionLocal();
+		String ip = conn.getHost();
+		String conf = conn.getScope().getName();
+		String inputLive = "rtmp://" + ip + "/video/" + conf + "/" + origin + " live=1";
+
+		String output = "rtmp://" + ip + "/video/" + conf + "/" + streamName;
+
+		ffmpeg = new FFmpegCommand();
+		ffmpeg.setFFmpegPath("/usr/local/bin/ffmpeg");
+		ffmpeg.setInput(inputLive);
+		ffmpeg.setFormat("flv");
+		ffmpeg.setOutput(output);
+		ffmpeg.setLoglevel("warning");
+		ffmpeg.setRotation(direction);
+
+		start();
+	}
+
+	/**
+	 * Get the stream name from the direction/streamName string
+	 * @param streamName Name of the stream with rotate direction
+	 * @return The stream name used for re-publish
+	 */
+	private String getStreamName(String streamName) {
+		String parts[] = streamName.split("/");
+		if(parts.length > 1)
+			return parts[1];
+		return "";
+	}
+
+	/**
+	 * Get the rotate direction from the streamName string.
+	 * @param streamName Name of the stream with rotate direction
+	 * @return FFmpegCommand.ROTATE for the given direction if present, null otherwise
+	 */
+	public static FFmpegCommand.ROTATE getDirection(String streamName) {
+		String parts[] = streamName.split("/");
+
+		switch(parts[0]) {
+			case ROTATE_LEFT:
+				return FFmpegCommand.ROTATE.LEFT;
+			case ROTATE_RIGHT:
+				return FFmpegCommand.ROTATE.RIGHT;
+			default:
+				return null;
+		}
+	}
+
+	/**
+	 * Start FFmpeg process to rotate and re-publish the stream.
+	 */
+	public void start() {
+		log.debug("Spawn FFMpeg to rotate [{}] stream [{}]", direction.name(), streamName);
+		String[] command = ffmpeg.getFFmpegCommand(true);
+		if (processMonitor == null) {
+			processMonitor = new ProcessMonitor(command);
+		}
+		processMonitor.start();
+	}
+
+	/**
+	 * Stop FFmpeg process that is rotating and re-publishing the stream.
+	 */
+	public void stop() {
+		log.debug("Stopping FFMpeg from rotate [{}] stream [{}]", direction.name(), streamName);
+		if(processMonitor != null) {
+			processMonitor.destroy();
+			processMonitor = null;
+		}
+	}
+}