diff --git a/.idea/libraries/imports.xml b/.idea/libraries/imports.xml index cacca7ef..c8db8146 100644 --- a/.idea/libraries/imports.xml +++ b/.idea/libraries/imports.xml @@ -31,6 +31,7 @@ + diff --git a/imports/async-http-client-2.0.18.jar b/imports/async-http-client-2.0.18.jar new file mode 100644 index 00000000..c4e563a9 Binary files /dev/null and b/imports/async-http-client-2.0.18.jar differ diff --git a/imports/httpasyncclient-4.1.2.jar b/imports/httpasyncclient-4.1.2.jar new file mode 100644 index 00000000..20a8a5a7 Binary files /dev/null and b/imports/httpasyncclient-4.1.2.jar differ diff --git a/imports/httpcore-nio-4.4.5.jar b/imports/httpcore-nio-4.4.5.jar new file mode 100644 index 00000000..18452862 Binary files /dev/null and b/imports/httpcore-nio-4.4.5.jar differ diff --git a/src/audio/broadcast/HttpAsyncAudioStreamer2.java b/src/audio/broadcast/HttpAsyncAudioStreamer2.java new file mode 100644 index 00000000..2aae8821 --- /dev/null +++ b/src/audio/broadcast/HttpAsyncAudioStreamer2.java @@ -0,0 +1,192 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2016 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * + ******************************************************************************/ +package audio.broadcast; + +import audio.AudioPacket; +import audio.IAudioPacketListener; +import audio.convert.IAudioConverter; +import org.apache.http.HttpException; +import org.apache.http.HttpHost; +import org.apache.http.HttpRequest; +import org.apache.http.HttpVersion; +import org.apache.http.client.methods.HttpPut; +import org.apache.http.entity.AbstractHttpEntity; +import org.apache.http.message.BasicHttpEntityEnclosingRequest; +import org.apache.http.nio.ContentEncoder; +import org.apache.http.nio.IOControl; +import org.apache.http.nio.entity.HttpAsyncContentProducer; +import org.apache.http.nio.protocol.HttpAsyncRequestProducer; +import org.apache.http.protocol.HttpContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sample.Listener; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; + +public class HttpAsyncAudioStreamer2 implements HttpAsyncRequestProducer, + IAudioPacketListener, Listener + +{ + private final static Logger mLog = LoggerFactory.getLogger( HttpAsyncAudioStreamer2.class ); + + private HttpHost mHttpHost; + private HttpRequest mHttpRequest; + private LinkedBlockingQueue mAudioQueue; + private IAudioConverter mIAudioConverter; + private IOControl mIOControl; + private AtomicBoolean mSuspended = new AtomicBoolean(); + + public HttpAsyncAudioStreamer2(HttpHost httpHost, HttpRequest httpRequest, + IAudioConverter audioConverter, int queueSize) + { + mHttpHost = httpHost; + mHttpRequest = httpRequest; + mIAudioConverter = audioConverter; + mAudioQueue = new LinkedBlockingQueue<>(queueSize); + } + + /** + * Queues the audio packet for conversion and streaming. Audio packets are assumed to arrive in sequence. + */ + @Override + public void receive(AudioPacket audioPacket) + { + //Insert the packet in the queue. If the queue is currently full, remove the head element and attempt + //to reinsert the packet. + if(!mAudioQueue.offer(audioPacket)) + { + mLog.debug("Streamer - making room to add an audio packet - suspended:" + mSuspended.get() + " iocontrol null:" + (mIOControl == null)); + mAudioQueue.poll(); + mAudioQueue.offer(audioPacket); + } + + if(mIOControl != null && mSuspended.compareAndSet(true, false)) + { + mLog.debug("Streamer - content available, resuming streaming"); + mIOControl.requestOutput(); + } + } + + @Override + public Listener getAudioPacketListener() + { + return this; + } + + + @Override + public HttpHost getTarget() + { + mLog.debug("Streamer - getTarget() invoked"); + return mHttpHost; + } + + @Override + public HttpRequest generateRequest() throws IOException, HttpException + { + mLog.debug("Streamer - generateRequest() invoked"); + return mHttpRequest; + } + + @Override + public void requestCompleted(HttpContext httpContext) + { + mLog.debug("Streamer - requestCompleted() invoked"); + } + + @Override + public void failed(Exception e) + { + mLog.error("Streamer - failed() invoked", e); + } + + @Override + public void resetRequest() throws IOException + { + mLog.debug("Streamer - resetRequest() invoked"); + } + + @Override + public void produceContent(ContentEncoder contentEncoder, IOControl ioControl) throws IOException + { + //IOControl is thread safe, so we'll keep a reference to it so that we can do flow control + if(mIOControl == null) + { + mIOControl = ioControl; + } + + AudioPacket packet = mAudioQueue.poll(); + + if(packet != null) + { + List packets = new ArrayList<>(); + packets.add(packet); + byte[] converted = mIAudioConverter.convert(packets); + mLog.debug("Produce Content - writing converted audio: " + converted.length + " to:" + contentEncoder.getClass()); + int wrote = contentEncoder.write(ByteBuffer.wrap(converted)); + + mLog.debug("We wrote [" + wrote + "] bytes to the content encoder"); + } + else + { + mLog.debug("Our audio packet was null"); + } + + if(mAudioQueue.isEmpty() && mSuspended.compareAndSet(false, true)) + { + mLog.debug("Produce Content - suspending output"); + mIOControl.suspendOutput(); + } + +// List packets = new ArrayList<>(); +// mAudioQueue.drainTo(packets); +// +// if(packets.isEmpty() && mSuspended.compareAndSet(false, true)) +// { +// mLog.debug("Produce Content - no audio - suspending output"); +// mIOControl.suspendOutput(); +// } +// else +// { +// byte[] converted = mIAudioConverter.convert(packets); +// mLog.debug("Produce Content - writing converted audio: " + converted.length); +// contentEncoder.write(ByteBuffer.wrap(converted)); +// } + } + + @Override + public boolean isRepeatable() + { + return false; + } + + @Override + public void close() throws IOException + { + mLog.debug("We were closed!"); + } + +} diff --git a/src/audio/broadcast/icecast/IcecastHandler.java b/src/audio/broadcast/icecast/IcecastHandler.java index 1161089f..35a78eda 100644 --- a/src/audio/broadcast/icecast/IcecastHandler.java +++ b/src/audio/broadcast/icecast/IcecastHandler.java @@ -24,12 +24,12 @@ import audio.broadcast.BroadcastHandler; import audio.broadcast.BroadcastState; import audio.broadcast.Broadcaster; import audio.broadcast.BroadcasterFactory; -import audio.broadcast.HttpAsyncAudioStreamer; +import audio.broadcast.HttpAsyncAudioStreamer2; import audio.convert.IAudioConverter; import controller.ThreadPoolManager; -import org.apache.http.HttpException; -import org.apache.http.HttpHeaders; +import org.apache.http.HttpEntity; import org.apache.http.HttpHost; +import org.apache.http.HttpRequest; import org.apache.http.HttpResponse; import org.apache.http.HttpStatus; import org.apache.http.HttpVersion; @@ -37,17 +37,16 @@ import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpPut; import org.apache.http.client.utils.URIBuilder; import org.apache.http.concurrent.FutureCallback; +import org.apache.http.entity.AbstractHttpEntity; import org.apache.http.entity.ContentType; -import org.apache.http.entity.InputStreamEntity; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; import org.apache.http.impl.nio.client.HttpAsyncClients; +import org.apache.http.nio.ContentDecoder; import org.apache.http.nio.IOControl; -import org.apache.http.nio.client.methods.AsyncCharConsumer; -import org.apache.http.nio.client.methods.HttpAsyncMethods; -import org.apache.http.nio.protocol.BasicAsyncRequestProducer; -import org.apache.http.nio.protocol.HttpAsyncRequestProducer; +import org.apache.http.nio.protocol.BasicAsyncResponseConsumer; +import org.apache.http.protocol.HTTP; import org.apache.http.protocol.HttpContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -55,11 +54,10 @@ import properties.SystemProperties; import record.wave.AudioPacketMonoWaveReader; import java.io.IOException; -import java.io.PipedInputStream; -import java.io.PipedOutputStream; +import java.io.InputStream; +import java.io.OutputStream; import java.net.URI; import java.net.URISyntaxException; -import java.nio.CharBuffer; import java.nio.file.Path; import java.nio.file.Paths; import java.util.List; @@ -73,8 +71,9 @@ public class IcecastHandler extends BroadcastHandler private static final long RECONNECT_INTERVAL_MILLISECONDS = 15000; //15 seconds private long mLastConnectionAttempt = 0; private CloseableHttpAsyncClient mHttpClient; - private HttpAsyncAudioStreamer mAudioStreamer; - private URI mURI; + private HttpAsyncAudioStreamer2 mAudioStreamer; + private HttpHost mHttpHost; + private HttpRequest mHttpRequest; private int mCounter = 0; /** @@ -86,8 +85,17 @@ public class IcecastHandler extends BroadcastHandler { super(configuration, audioConverter); - mAudioStreamer = new HttpAsyncAudioStreamer(HttpPut.METHOD_NAME, getConfiguration().getMountPoint(), - getAudioConverter(), 20); + try + { + HttpRequest httpRequest = getHttpRequest(); + mAudioStreamer = new HttpAsyncAudioStreamer2(getHttpHost(), httpRequest, + getAudioConverter(), 20); + } + catch(URISyntaxException use) + { + mLog.error("Error creating URI from configuration", use); + setBroadcastState(BroadcastState.UNKNOWN_HOST); + } } /** @@ -98,19 +106,49 @@ public class IcecastHandler extends BroadcastHandler return (IcecastConfiguration)getBroadcastConfiguration(); } - private URI getURI() throws URISyntaxException + private HttpHost getHttpHost() { - if(mURI == null) + if(mHttpHost == null) { - mURI = new URIBuilder() - .setScheme("http") - .setHost(getConfiguration().getHost()) - .setPort(getConfiguration().getPort()) - .setPath(getConfiguration().getMountPoint()) - .build(); + mHttpHost = new HttpHost(getConfiguration().getHost(), getConfiguration().getPort()); } - return mURI; + return mHttpHost; + } + + private URI getURI() throws URISyntaxException + { + return new URIBuilder() + .setScheme("http") + .setHost(getConfiguration().getHost()) + .setPort(getConfiguration().getPort()) + .setPath(getConfiguration().getMountPoint()) + .build(); + } + + private HttpRequest getHttpRequest() throws URISyntaxException + { + if(mHttpRequest == null) + { + HttpPut httpPut = new HttpPut(getURI()); + httpPut.setProtocolVersion(HttpVersion.HTTP_1_1); + + httpPut.setEntity(new MyHttpEntity(getConfiguration().getBroadcastFormat())); + + mHttpRequest = httpPut; + mHttpRequest.addHeader(IcecastHeader.AUTHORIZATION.getValue(), getConfiguration().getAuthorization()); + mHttpRequest.addHeader(IcecastHeader.ACCEPT.getValue(), "*/*"); + mHttpRequest.addHeader(IcecastHeader.CONTENT_TYPE.getValue(), getConfiguration().getBroadcastFormat().getValue()); + mHttpRequest.addHeader(IcecastHeader.PUBLIC.getValue(), getConfiguration().isPublic() ? "1" : "0"); + mHttpRequest.addHeader(IcecastHeader.NAME.getValue(), getConfiguration().getStreamName()); + mHttpRequest.addHeader(IcecastHeader.DESCRIPTION.getValue(), getConfiguration().getHost()); + mHttpRequest.addHeader(IcecastHeader.URL.getValue(), getConfiguration().getURL()); + mHttpRequest.addHeader(IcecastHeader.GENRE.getValue(), getConfiguration().getGenre()); + mHttpRequest.addHeader(IcecastHeader.USER_AGENT.getValue(), SystemProperties.getInstance().getApplicationName()); + mHttpRequest.addHeader(HTTP.EXPECT_DIRECTIVE, HTTP.EXPECT_CONTINUE); + } + + return mHttpRequest; } /** @@ -120,7 +158,7 @@ public class IcecastHandler extends BroadcastHandler @Override public void broadcast(List audioPackets) { - if(mCounter > 10) + if(mCounter > 1) { mLog.debug("Connecting ..."); connect(); @@ -179,90 +217,87 @@ public class IcecastHandler extends BroadcastHandler // try // { - HttpHost httpHost = new HttpHost(getConfiguration().getHost(), getConfiguration().getPort()); // mAudioStreamer.addHeader(IcecastHeader.EXPECT.getValue(), HTTP_100_CONTINUE); - mAudioStreamer.addHeader(IcecastHeader.AUTHORIZATION.getValue(), getConfiguration().getAuthorization()); - mAudioStreamer.addHeader(HttpHeaders.ACCEPT, "*/*"); - mAudioStreamer.addHeader(HttpHeaders.CONTENT_TYPE, getConfiguration().getBroadcastFormat().getValue()); - mAudioStreamer.addHeader(IcecastHeader.PUBLIC.getValue(), getConfiguration().isPublic() ? "1" : "0"); - mAudioStreamer.addHeader(IcecastHeader.NAME.getValue(), getConfiguration().getStreamName()); - mAudioStreamer.addHeader(IcecastHeader.DESCRIPTION.getValue(), getConfiguration().getHost()); - mAudioStreamer.addHeader(IcecastHeader.URL.getValue(), getConfiguration().getURL()); - mAudioStreamer.addHeader(IcecastHeader.GENRE.getValue(), getConfiguration().getGenre()); - mAudioStreamer.addHeader(IcecastHeader.USER_AGENT.getValue(), SystemProperties.getInstance().getApplicationName()); // BasicAsyncRequestProducer producer = new BasicAsyncRequestProducer(httpHost, mAudioStreamer); - BasicAsyncRequestProducer producer = new MyProducer(httpHost, mAudioStreamer, mAudioStreamer); - AsyncCharConsumer consumer = new AsyncCharConsumer() - { - @Override - protected void onCharReceived(CharBuffer charBuffer, IOControl ioControl) throws IOException - { - mLog.debug("Received consumer characters!"); - } +// BasicAsyncRequestProducer producer = new MyProducer(httpHost, mAudioStreamer, mAudioStreamer); +// HttpAsyncRequestProducer producer = new HttpAsyncAudioStreamer2(httpHost, getConfiguration().getBroadcastFormat(), +// getAudioConverter(), 2000); - @Override - protected void onResponseReceived(HttpResponse response) throws HttpException, IOException - { - mLog.debug("Received consumer response:\n" + response.toString()); - switch(response.getStatusLine().getStatusCode()) - { - case HttpStatus.SC_OK: //200 - case HttpStatus.SC_CONTINUE: //100 - mLog.debug("Connected to icecast server"); - setBroadcastState(BroadcastState.CONNECTED); - break; - case HttpStatus.SC_UNAUTHORIZED: //401 - setBroadcastState(BroadcastState.INVALID_PASSWORD); - mLog.error("Couldn't connect to Icecast2 server using login credentials"); - break; - case HttpStatus.SC_FORBIDDEN: //403 - String reason = response.getStatusLine().getReasonPhrase(); +// AsyncCharConsumer consumer = new AsyncCharConsumer() +// { +// @Override +// protected void onCharReceived(CharBuffer charBuffer, IOControl ioControl) throws IOException +// { +// mLog.debug("Received consumer characters!"); +// } +// +// @Override +// protected void onResponseReceived(HttpResponse response) throws HttpException, IOException +// { +// mLog.debug("Received consumer response:\n" + response.toString()); +// +// switch(response.getStatusLine().getStatusCode()) +// { +// case HttpStatus.SC_OK: //200 +// case HttpStatus.SC_CONTINUE: //100 +// mLog.debug("Connected to icecast server"); +// setBroadcastState(BroadcastState.CONNECTED); +// break; +// case HttpStatus.SC_UNAUTHORIZED: //401 +// setBroadcastState(BroadcastState.INVALID_PASSWORD); +// mLog.error("Couldn't connect to Icecast2 server using login credentials"); +// break; +// case HttpStatus.SC_FORBIDDEN: //403 +// String reason = response.getStatusLine().getReasonPhrase(); +// +// if(reason.contains("Content-Type")) +// { +// setBroadcastState(BroadcastState.UNSUPPORTED_AUDIO_FORMAT); +// mLog.error("Icecast2 server does not support audio format"); +// } +// else if(reason.startsWith("internal format")) +// { +// setBroadcastState(BroadcastState.REMOTE_SERVER_ERROR); +// mLog.error("Icecast2 server has an error"); +// } +// else if(reason.startsWith("too many sources")) +// { +// setBroadcastState(BroadcastState.MAX_SOURCES_EXCEEDED); +// mLog.error("Icecast2 server maximum number of sources exceeded"); +// } +// else if(reason.startsWith("Mountpoint")) +// { +// setBroadcastState(BroadcastState.MOUNT_POINT_IN_USE); +// mLog.error("Icecast2 server mount point is already in use"); +// } +// case HttpStatus.SC_INTERNAL_SERVER_ERROR: //500 +// setBroadcastState(BroadcastState.INVALID_PASSWORD); +// mLog.error("Couldn't connect to Icecast2 server using login credentials"); +// break; +// default: +// setBroadcastState(BroadcastState.ERROR); +// mLog.error("Error connecting to remote server. HTTP Status Code:" + +// response.getStatusLine().getStatusCode() + " Reason:" + +// response.getStatusLine().getReasonPhrase()); +// } +// +// } +// +// @Override +// protected HttpResponse buildResult(HttpContext httpContext) throws Exception +// { +// mLog.debug("Received consumer request for response!!!!!!"); +// return null; +// } +// }; - if(reason.contains("Content-Type")) - { - setBroadcastState(BroadcastState.UNSUPPORTED_AUDIO_FORMAT); - mLog.error("Icecast2 server does not support audio format"); - } - else if(reason.startsWith("internal format")) - { - setBroadcastState(BroadcastState.REMOTE_SERVER_ERROR); - mLog.error("Icecast2 server has an error"); - } - else if(reason.startsWith("too many sources")) - { - setBroadcastState(BroadcastState.MAX_SOURCES_EXCEEDED); - mLog.error("Icecast2 server maximum number of sources exceeded"); - } - else if(reason.startsWith("Mountpoint")) - { - setBroadcastState(BroadcastState.MOUNT_POINT_IN_USE); - mLog.error("Icecast2 server mount point is already in use"); - } - case HttpStatus.SC_INTERNAL_SERVER_ERROR: //500 - setBroadcastState(BroadcastState.INVALID_PASSWORD); - mLog.error("Couldn't connect to Icecast2 server using login credentials"); - break; - default: - setBroadcastState(BroadcastState.ERROR); - mLog.error("Error connecting to remote server. HTTP Status Code:" + - response.getStatusLine().getStatusCode() + " Reason:" + - response.getStatusLine().getReasonPhrase()); - } - } - - @Override - protected HttpResponse buildResult(HttpContext httpContext) throws Exception - { - mLog.debug("Received consumer request for response!!!!!!"); - return null; - } - }; + MyConsumer consumer = new MyConsumer(); mLog.debug("Sending connection request"); - Future response = mHttpClient.execute(producer, consumer, new FutureCallback() + Future response = mHttpClient.execute(mAudioStreamer, consumer, new FutureCallback() { @Override public void completed(HttpResponse httpResponse) @@ -378,4 +413,162 @@ public class IcecastHandler extends BroadcastHandler } } } + + + public class MyHttpEntity extends AbstractHttpEntity + { + public MyHttpEntity(BroadcastFormat broadcastFormat) + { + super.setContentType(broadcastFormat.getValue()); + super.setChunked(false); + } + + @Override + public boolean isRepeatable() + { + mLog.debug("MyEntity - isRepeatable() invoked"); + return false; + } + + @Override + public long getContentLength() + { + return -1; + } + + @Override + public InputStream getContent() throws IOException, UnsupportedOperationException + { + mLog.debug("MyEntity - getContent() invoked"); + return null; + } + + @Override + public void writeTo(OutputStream outputStream) throws IOException + { + mLog.debug("MyEntity - writeTo() invoked"); + } + + @Override + public boolean isStreaming() + { + mLog.debug("MyEntity - isStreaming() invoked"); + return true; + } + } + + public class MyConsumer extends BasicAsyncResponseConsumer + { + public MyConsumer() + { + + } + + @Override + protected void onResponseReceived(HttpResponse response) throws IOException + { + mLog.debug("MyConsumer onResponseReceived() " + response.toString()); + + super.onResponseReceived(response); + + switch(response.getStatusLine().getStatusCode()) + { + case HttpStatus.SC_OK: //200 + case HttpStatus.SC_CONTINUE: //100 + mLog.debug("Connected to icecast server"); + setBroadcastState(BroadcastState.CONNECTED); + break; + case HttpStatus.SC_UNAUTHORIZED: //401 + setBroadcastState(BroadcastState.INVALID_PASSWORD); + mLog.error("Couldn't connect to Icecast2 server using login credentials"); + break; + case HttpStatus.SC_FORBIDDEN: //403 + String reason = response.getStatusLine().getReasonPhrase(); + + if(reason.contains("Content-Type")) + { + setBroadcastState(BroadcastState.UNSUPPORTED_AUDIO_FORMAT); + mLog.error("Icecast2 server does not support audio format"); + } + else if(reason.startsWith("internal format")) + { + setBroadcastState(BroadcastState.REMOTE_SERVER_ERROR); + mLog.error("Icecast2 server has an error"); + } + else if(reason.startsWith("too many sources")) + { + setBroadcastState(BroadcastState.MAX_SOURCES_EXCEEDED); + mLog.error("Icecast2 server maximum number of sources exceeded"); + } + else if(reason.startsWith("Mountpoint")) + { + setBroadcastState(BroadcastState.MOUNT_POINT_IN_USE); + mLog.error("Icecast2 server mount point is already in use"); + } + case HttpStatus.SC_INTERNAL_SERVER_ERROR: //500 + setBroadcastState(BroadcastState.INVALID_PASSWORD); + mLog.error("Couldn't connect to Icecast2 server using login credentials"); + break; + default: + setBroadcastState(BroadcastState.ERROR); + mLog.error("Error connecting to remote server. HTTP Status Code:" + + response.getStatusLine().getStatusCode() + " Reason:" + + response.getStatusLine().getReasonPhrase()); + } + } + + @Override + protected void onEntityEnclosed(HttpEntity entity, ContentType contentType) throws IOException + { + mLog.debug("MyConsumer - onEntityEnclosed()"); + super.onEntityEnclosed(entity, contentType); + } + + @Override + protected void onContentReceived(ContentDecoder decoder, IOControl ioctrl) throws IOException + { + mLog.debug("MyConsumer - onContentReceived()"); + super.onContentReceived(decoder, ioctrl); + + mLog.debug("MyConsumer - requesting output from the IO control"); + ioctrl.requestOutput(); + } + + @Override + protected HttpResponse buildResult(HttpContext context) + { + mLog.debug("MyConsumer - buildResult()"); + return super.buildResult(context); + } + + @Override + protected void onClose() throws IOException + { + mLog.debug("MyConsumer - onClose()"); + super.onClose(); + } + + @Override + public Exception getException() + { + mLog.debug("MyConsumer - getException()"); + return super.getException(); + } + + @Override + public HttpResponse getResult() + { + mLog.debug("MyConsumer - getResult()"); + return super.getResult(); + } + + @Override + public boolean isDone() + { + boolean done = super.isDone(); + + mLog.debug("MyConsumer - isDone() - super says:" + done); + return super.isDone(); + } + } }