Fix leak socket

src/main/java/io/antmedia/muxer/RtmpMuxer.java

@@ -53,6 +53,7 @@ public class RtmpMuxer extends Muxer {
 	boolean keyFrameReceived = false;
 
 	private AtomicBoolean preparedIO = new AtomicBoolean(false);
+	private AtomicBoolean cancelOpenIO = new AtomicBoolean(false);
 
 	public RtmpMuxer(String url, Vertx vertx) {
 		super(vertx);
@@ -141,32 +142,34 @@ public synchronized boolean prepareIO()
 			return false;
 		}
 		preparedIO.set(true);
+		cancelOpenIO.set(false);
 		boolean result = false;
 		//if there is a stream in the output format context, try to push
 		if (getOutputFormatContext().nb_streams() > 0) 
 		{
 			this.vertx.executeBlocking(() -> {
-
 				if (openIO())
 				{
 					if (bsfFilterContextList.isEmpty())
 					{
 						writeHeader();
 						return null;
 					}
-					isRunning.set(true);
-					setStatus(IAntMediaStreamHandler.BROADCAST_STATUS_BROADCASTING);
+					if (!exitIfCancelled())
+					{
+						isRunning.set(true);
+						setStatus(IAntMediaStreamHandler.BROADCAST_STATUS_BROADCASTING);
+					}
+
 				}
 				else
 				{
 					clearResource();
 					setStatus(IAntMediaStreamHandler.BROADCAST_STATUS_FAILED);
 					logger.error("Cannot initializeOutputFormatContextIO for rtmp endpoint:{}", url);
 				}
-				
+
 				return null;
-
-
 			}, false);
 
 			result = true;
@@ -210,19 +213,24 @@ public synchronized boolean writeHeader() {
 	 */
 	@Override
 	public synchronized void writeTrailer() {
+		cancelOpenIO.set(true);
 		if(headerWritten){
 			super.writeTrailer();
 			trailerWritten = true;
 		}
 		else{
 			logger.info("Not writing trailer because header is not written yet");
+			clearResource();
 		}
 		setStatus(IAntMediaStreamHandler.BROADCAST_STATUS_FINISHED);
 	}
 
 	@Override
 	public synchronized void clearResource() {
 		super.clearResource();
+		if (!headerWritten) {
+			preparedIO.set(false);
+		}
 		/**
 		 *  Don't free the allocatedExtraDataPointer because it's internally deallocated
 		 *
@@ -235,6 +243,15 @@ public synchronized void clearResource() {
 		//allocatedExtraDataPointer is freed when the context is closing
 	}
 
+	private boolean exitIfCancelled() {
+		if (!cancelOpenIO.get()) {
+			return false;
+		}
+		logger.info("RTMP muxer openIO cancelled for {}", url);
+		clearResource();
+		return true;
+	}
+
 	/**
 	 * {@inheritDoc}
 	 */