GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-12 16:47:31 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

#30 WIP extracted the IcecastMetadataUpdater and merged some of the icecast configuration method into a new superclass.

Browse Source
This commit is contained in:
Denny
2016-12-26 09:08:34 -05:00
parent c216f56c4f
commit 524a76e583
29 changed files with 655 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
.idea/libraries/org_asynchttpclient_async_http_client_2_0_18.xml generated
View File

@@ -3,16 +3,7 @@
<properties maven-id="org.asynchttpclient:async-http-client:2.0.18" />
<CLASSES>
<root url="jar://$PROJECT_DIR$/imports/slf4j-api-1.7.21.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/reactive-streams-1.0.0.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-reactive-streams-1.0.8.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/javassist-3.20.0-GA.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-buffer-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-codec-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-codec-http-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-common-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-handler-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-transport-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-transport-native-epoll-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/axis.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/jaxrpc.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/commons-discovery-0.2.jar!/" />
@@ -20,11 +11,6 @@
<root url="jar://$PROJECT_DIR$/imports/java-mail-1.4.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/wsdl4j-1.5.1.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/radioreference-14.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-codec-dns-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-resolver-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/netty-resolver-dns-4.1.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/async-http-client-2.0.24.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/async-http-client-netty-utils-2.0.24.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/mina-core-2.0.16.jar!/" />
<root url="jar://$PROJECT_DIR$/imports/mina-http-2.0.16.jar!/" />
</CLASSES>

3
config/logback.xml
View File

@@ -30,6 +30,9 @@
<!-- Async HTTP Client - set level to info to filter debug messages -->
<logger name="org.asynchttpclient.netty" level="INFO" />
<!-- Async HTTP Client - set level to info to filter debug messages -->
<logger name="org.apache.mina" level="INFO" />
<!-- IO Netty - set level to info to filter debug messages -->
<logger name="io.netty" level="INFO" />

BIN
imports/async-http-client-2.0.24.jar
View File

Binary file not shown.

BIN
imports/netty-buffer-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-codec-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-codec-dns-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-codec-http-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-common-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-handler-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-reactive-streams-1.0.8.jar
View File

Binary file not shown.

BIN
imports/netty-resolver-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-resolver-dns-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-transport-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/netty-transport-native-epoll-4.1.6.Final.jar
View File

Binary file not shown.

BIN
imports/reactive-streams-1.0.0.jar
View File

Binary file not shown.

14
src/audio/broadcast/AudioBroadcaster.java
View File

@@ -190,8 +190,11 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
*/
public void receive(StreamableAudioRecording recording)
{
mAudioRecordingQueue.add(recording);
broadcast(new BroadcastEvent(this, BroadcastEvent.Event.BROADCASTER_QUEUE_CHANGE));
if(connected())
{
mAudioRecordingQueue.add(recording);
broadcast(new BroadcastEvent(this, BroadcastEvent.Event.BROADCASTER_QUEUE_CHANGE));
}
}
/**
@@ -262,7 +265,7 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
{
if(mBroadcastState != state)
{
mLog.debug("[" + getStreamName() + "] changing state to: " + state);
mLog.info("[" + getStreamName() + "] changing state to: " + state);
mBroadcastState = state;
broadcast(new BroadcastEvent(this, BroadcastEvent.Event.BROADCASTER_STATE_CHANGE));
@@ -271,6 +274,11 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
{
stop();
}
if(!connected())
{
mAudioRecordingQueue.clear();
}
}
}

21
src/audio/broadcast/BroadcastFactory.java
View File

@@ -28,16 +28,15 @@ import audio.broadcast.icecast.IcecastHTTPConfigurationEditor;
import audio.broadcast.icecast.IcecastTCPAudioBroadcaster;
import audio.broadcast.icecast.IcecastTCPConfiguration;
import audio.broadcast.icecast.IcecastTCPConfigurationEditor;
import audio.broadcast.shoutcast.v1.ShoutcastV1Configuration;
import audio.broadcast.shoutcast.v1.ShoutcastV1AudioBroadcaster;
import audio.broadcast.shoutcast.v2.ShoutcastV2Configuration;
import audio.broadcast.shoutcast.v1.ShoutcastV1Configuration;
import audio.broadcast.shoutcast.v2.ShoutcastV2AudioBroadcaster;
import audio.broadcast.shoutcast.v2.ShoutcastV2Configuration;
import audio.convert.IAudioConverter;
import audio.convert.MP3AudioConverter;
import controller.ThreadPoolManager;
import gui.editor.Editor;
import gui.editor.EmptyEditor;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import record.AudioRecorder;
@@ -54,7 +53,6 @@ public class BroadcastFactory
public static final int MP3_MONO_16_KHZ_BITRATE = 16;
public static final boolean MP3_CONSTANT_BITRATE = false;
public static final boolean MP3_VARIABLE_BITRATE = true;
/**
* Creates an audio streaming broadcaster for the configuration
@@ -62,8 +60,7 @@ public class BroadcastFactory
* @param configuration describing the server and audio types
* @return configured broadcaster or null
*/
public static AudioBroadcaster getBroadcaster(DefaultAsyncHttpClient defaultAsyncHttpClient,
ThreadPoolManager threadPoolManager,
public static AudioBroadcaster getBroadcaster(ThreadPoolManager threadPoolManager,
BroadcastConfiguration configuration)
{
if(configuration != null)
@@ -75,18 +72,20 @@ public class BroadcastFactory
switch(configuration.getBroadcastServerType())
{
case BROADCASTIFY:
return new IcecastTCPAudioBroadcaster(defaultAsyncHttpClient, threadPoolManager,
return new IcecastTCPAudioBroadcaster(threadPoolManager,
(BroadcastifyConfiguration)configuration);
case ICECAST_TCP:
return new IcecastTCPAudioBroadcaster(defaultAsyncHttpClient, threadPoolManager,
return new IcecastTCPAudioBroadcaster(threadPoolManager,
(IcecastTCPConfiguration)configuration);
case ICECAST_HTTP:
return new IcecastHTTPAudioBroadcaster(defaultAsyncHttpClient, threadPoolManager,
return new IcecastHTTPAudioBroadcaster(threadPoolManager,
(IcecastHTTPConfiguration)configuration);
case SHOUTCAST_V1:
return new ShoutcastV1AudioBroadcaster(threadPoolManager, (ShoutcastV1Configuration)configuration);
return new ShoutcastV1AudioBroadcaster(threadPoolManager,
(ShoutcastV1Configuration)configuration);
case SHOUTCAST_V2:
return new ShoutcastV2AudioBroadcaster(threadPoolManager, (ShoutcastV2Configuration)configuration);
return new ShoutcastV2AudioBroadcaster(threadPoolManager,
(ShoutcastV2Configuration)configuration);
case UNKNOWN:
default:
mLog.info("Unrecognized broadcastAudio configuration: " + configuration.getBroadcastFormat().name());

38
src/audio/broadcast/BroadcastModel.java
View File

@@ -21,8 +21,6 @@ package audio.broadcast;
import alias.id.broadcast.BroadcastChannel;
import audio.AudioPacket;
import controller.ThreadPoolManager;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClientConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import properties.SystemProperties;
@@ -65,7 +63,6 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
private Map<String,BroadcastConfiguration> mBroadcastConfigurationMap = new HashMap<>();
private Map<String,AudioBroadcaster> mBroadcasterMap = new HashMap<>();
private DefaultAsyncHttpClient mDefaultAsyncHttpClient;
private ThreadPoolManager mThreadPoolManager;
private SettingsManager mSettingsManager;
@@ -297,8 +294,8 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
broadcastConfiguration.isValid() &&
!mBroadcasterMap.containsKey(broadcastConfiguration.getName()))
{
AudioBroadcaster audioBroadcaster = BroadcastFactory.getBroadcaster(getDefaultAsyncHttpClient(),
mThreadPoolManager, broadcastConfiguration);
AudioBroadcaster audioBroadcaster = BroadcastFactory.getBroadcaster(mThreadPoolManager,
broadcastConfiguration);
if (audioBroadcaster != null)
{
@@ -353,37 +350,6 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
}
}
/**
* HTTP client shared across all broadcaster instances. Note: client is lazy construct so that it isn't
* created until required
*/
private DefaultAsyncHttpClient getDefaultAsyncHttpClient()
{
if (mDefaultAsyncHttpClient == null)
{
//Set read timeout to never expire so that our streams don't get interrupted
DefaultAsyncHttpClientConfig config =
new DefaultAsyncHttpClientConfig.Builder()
.setReadTimeout(-1)
.setThreadPoolName("audio-broadcast-pool")
.setUserAgent(SystemProperties.getInstance().getApplicationName())
.build();
mLog.debug("Client Config - Request Timeout:" + config.getRequestTimeout());
mLog.debug("Client Config - Connection Ttl:" + config.getConnectionTtl());
mLog.debug("Client Config - Connect Timeout:" + config.getConnectTimeout());
mLog.debug("Client Config - IO Thread Count:" + config.getIoThreadsCount());
mLog.debug("Client Config - Pool Conn Idle Timeout:" + config.getPooledConnectionIdleTimeout());
mLog.debug("Client Config - Connection Pooling:" + config.getIoThreadsCount());
mDefaultAsyncHttpClient = new DefaultAsyncHttpClient(config);
}
return mDefaultAsyncHttpClient;
}
/**
* Returns the broadcast configuration identified by the stream name
*/

17
src/audio/broadcast/BroadcastState.java
View File

@@ -21,9 +21,9 @@ package audio.broadcast;
public enum BroadcastState
{
/**
* Error while broadcasting stream data. Temporary error state to allow connection to be reset.
* General error in configuration that causes the remote server to reject the connection
*/
BROADCAST_ERROR("Broadcast Error", false),
CONFIGURATION_ERROR("Configuration Error", true),
/**
* Connected to the broadcastAudio server and capable of streaming
@@ -41,9 +41,9 @@ public enum BroadcastState
DISCONNECTED("Disconnected", false),
/**
* Invalid password
* Invalid credentials - user name or password
*/
INVALID_PASSWORD("Invalid Password", true),
INVALID_CREDENTIALS("Invalid User Name/Password", true),
/**
* Remote server max sources has been exceeded
@@ -66,14 +66,9 @@ public enum BroadcastState
READY("Ready", false),
/**
* Error on the remote server
* Error while broadcasting stream data. Temporary error state to allow connection to be reset.
*/
REMOTE_SERVER_ERROR("Remote Server Error", true),
/**
* Server host name or port is invalid
*/
UNKNOWN_HOST("Unknown Host or Invalid Port", true),
TEMPORARY_BROADCAST_ERROR("Temporary Broadcast Error", false),
/**
* Unsupported audio format

56
src/audio/broadcast/icecast/IcecastAudioBroadcaster.java Normal file
View File

@@ -0,0 +1,56 @@
/*******************************************************************************
* 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 <http://www.gnu.org/licenses/>
*
******************************************************************************/
package audio.broadcast.icecast;
import audio.broadcast.AudioBroadcaster;
import audio.broadcast.BroadcastConfiguration;
import audio.metadata.AudioMetadata;
import controller.ThreadPoolManager;
public abstract class IcecastAudioBroadcaster extends AudioBroadcaster
{
private IcecastMetadataUpdater mIcecastMetadataUpdater;
public IcecastAudioBroadcaster(ThreadPoolManager threadPoolManager, BroadcastConfiguration broadcastConfiguration)
{
super(threadPoolManager, broadcastConfiguration);
}
/**
* Icecast broadcast configuration
*/
protected IcecastConfiguration getConfiguration()
{
return (IcecastConfiguration)getBroadcastConfiguration();
}
/**
* Broadcasts an audio metadata update
*/
@Override
protected void broadcastMetadata(AudioMetadata metadata)
{
if(mIcecastMetadataUpdater == null)
{
mIcecastMetadataUpdater = new IcecastMetadataUpdater(getThreadPoolManager(), getConfiguration());
}
mIcecastMetadataUpdater.update(metadata);
}
}

562
src/audio/broadcast/icecast/IcecastHTTPAudioBroadcaster.java
View File

@@ -18,212 +18,152 @@
******************************************************************************/
package audio.broadcast.icecast;
import audio.AudioPacket;
import audio.broadcast.AudioBroadcaster;
import audio.broadcast.BroadcastState;
import audio.metadata.AudioMetadata;
import audio.metadata.Metadata;
import audio.metadata.MetadataType;
import audio.convert.MP3AudioConverter;
import controller.ThreadPoolManager;
import org.asynchttpclient.AsyncHandler;
import org.asynchttpclient.BoundRequestBuilder;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.HttpResponseBodyPart;
import org.asynchttpclient.HttpResponseHeaders;
import org.asynchttpclient.HttpResponseStatus;
import org.asynchttpclient.ListenableFuture;
import org.asynchttpclient.Realm;
import org.asynchttpclient.Response;
import org.asynchttpclient.uri.Uri;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.apache.mina.core.RuntimeIoException;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolDecoderException;
import org.apache.mina.http.HttpClientCodec;
import org.apache.mina.http.HttpException;
import org.apache.mina.http.HttpRequestImpl;
import org.apache.mina.http.api.DefaultHttpResponse;
import org.apache.mina.http.api.HttpMethod;
import org.apache.mina.http.api.HttpStatus;
import org.apache.mina.http.api.HttpVersion;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import properties.SystemProperties;
import java.net.ConnectException;
import java.net.URLEncoder;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
public class IcecastHTTPAudioBroadcaster extends AudioBroadcaster implements Publisher<ByteBuffer>
public class IcecastHTTPAudioBroadcaster extends IcecastAudioBroadcaster
{
private final static Logger mLog = LoggerFactory.getLogger( IcecastHTTPAudioBroadcaster.class );
private static final Logger mLog = LoggerFactory.getLogger( IcecastHTTPAudioBroadcaster.class );
private static final long RECONNECT_INTERVAL_MILLISECONDS = 30000; //30 seconds
private NioSocketConnector mSocketConnector;
private IoSession mStreamingSession = null;
private Map<String, String> mHTTPHeaders;
private static final String UTF8 = "UTF-8";
private static final String HTTP_100_CONTINUE = "100-continue";
private static final long RECONNECT_INTERVAL_MILLISECONDS = 15000; //15 seconds
private long mLastConnectionAttempt = 0;
private DefaultAsyncHttpClient mDefaultAsyncHttpClient;
private Subscriber<? super ByteBuffer> mSubscriber;
private List<AudioPacket> mPacketsToBroadcast = new ArrayList<>();
private ConvertedAudioStreamSubscription mSubscription;
private AtomicBoolean mConnecting = new AtomicBoolean();
private boolean mFirstMetadataUpdateSuppressed = false;
/**
* Creates an Icecast 2 (v2.4.x or newer) compatible broadcaster using HTTP 1.1 protocol
* Creates an Icecast 2.3.2 compatible broadcaster using TCP and a pseudo HTTP 1.0 protocol. This broadcaster is
* compatible with Icecast version 2.3.2 and older versions of the server software.
*
* Note: use @see IcecastTCPAudioBroadcaster for Icecast 1 (v2.3.x and older).
* Note: use @see IcecastHTTPAudioBroadcaster for Icecast version 2.4.x and newer.
*
* @param defaultAsyncHttpClient to use when communicating with a remote HTTP server
* @param threadPoolManager
* @param configuration to use when connecting to the remote server
* This broadcaster uses the Apache Mina library for the streaming socket connection and for metadata updates. The
* IcecastHTTPIOHandler manages all interaction with the Icecast server and manages the overall broadcast state.
*
* @param configuration for the Icecast stream
*/
public IcecastHTTPAudioBroadcaster(DefaultAsyncHttpClient defaultAsyncHttpClient,
ThreadPoolManager threadPoolManager,
IcecastHTTPConfiguration configuration)
public IcecastHTTPAudioBroadcaster(ThreadPoolManager threadPoolManager, IcecastHTTPConfiguration configuration)
{
super(threadPoolManager, configuration);
mDefaultAsyncHttpClient = defaultAsyncHttpClient;
}
/**
* Configuration information
*/
private IcecastHTTPConfiguration getConfiguration()
{
return (IcecastHTTPConfiguration)getBroadcastConfiguration();
}
@Override
protected void broadcastMetadata(AudioMetadata metadata)
{
if(connected() && mFirstMetadataUpdateSuppressed)
{
StringBuilder sb = new StringBuilder();
if(metadata != null)
{
Metadata to = metadata.getMetadata(MetadataType.TO);
sb.append("TO:");
if(to != null)
{
if(to.hasAlias())
{
sb.append(to.getAlias().getName());
}
else
{
sb.append(to.getValue());
}
}
else
{
sb.append("UNKNOWN");
}
Metadata from = metadata.getMetadata(MetadataType.FROM);
sb.append(" FROM:");
if(from != null)
{
if(from.hasAlias())
{
sb.append(from.getAlias().getName());
}
else
{
sb.append(from.getValue());
}
}
else
{
sb.append("UNKNOWN");
}
}
else
{
sb.append("Scanning ....");
}
try
{
if(mDefaultAsyncHttpClient == null)
{
mDefaultAsyncHttpClient = new DefaultAsyncHttpClient();
}
String songEncoded = URLEncoder.encode(sb.toString(), UTF8);
StringBuilder query = new StringBuilder();
query.append("/admin/metadata?mode=updinfo");
query.append("&mount=").append(getConfiguration().getMountPoint());
query.append("&charset=UTF%2d8");
query.append("&song=").append(songEncoded);
Uri uri = new Uri(Uri.HTTP, null, getConfiguration().getHost(), getConfiguration().getPort(),
query.toString(), null);
BoundRequestBuilder builder = mDefaultAsyncHttpClient.prepareGet(uri.toUrl());
//Use Basic (base64) authentication
Realm realm = new Realm.Builder(getConfiguration().getUserName(), getConfiguration().getPassword())
.setScheme(Realm.AuthScheme.BASIC).setUsePreemptiveAuth(true).build();
builder.setRealm(realm);
builder.addHeader(IcecastHeader.USER_AGENT.getValue(), SystemProperties.getInstance().getApplicationName());
mDefaultAsyncHttpClient.executeRequest(builder.build());
}
catch(Exception e)
{
mLog.debug("Error while updating metadata", e);
}
}
else if(connected())
{
mFirstMetadataUpdateSuppressed = true;
}
}
/**
* Broadcast converted audio chunks received from the parent class processor
* Broadcasts the audio frame or sequence
*/
@Override
protected void broadcastAudio(byte[] audio)
{
if(connect())
{
if (mSubscription != null && mSubscription.hasDemand())
if(mStreamingSession != null && mStreamingSession.isConnected())
{
mSubscription.broadcast(ByteBuffer.wrap(audio));
IoBuffer buffer = IoBuffer.allocate(audio.length);
buffer.put(audio);
buffer.flip();
mStreamingSession.write(buffer);
}
}
}
@Override
public void subscribe(Subscriber<? super ByteBuffer> subscriber)
{
//Remove existing subscription, if there is one
if (mSubscription != null)
{
mSubscription.cancel();
}
//Create the subscription
mSubscription = new ConvertedAudioStreamSubscription(subscriber);
}
/**
* Indicates if the audio handler is currently connected to the remote server and capable of streaming audio.
* (Re)Connects the broadcaster to the remote server if it currently is disconnected and indicates if the broadcaster
* is currently connected to the remote server following any connection attempts.
*
* Attempts to connect via this method when the broadcast state indicates an error condition will be ignored.
*
* This method will attempt to establish a connection or reconnect to the streaming server.
* @return true if the audio handler can stream audio
*/
private boolean connect()
{
if(canConnect())
if(!connected() && canConnect() &&
(mLastConnectionAttempt + RECONNECT_INTERVAL_MILLISECONDS < System.currentTimeMillis()) &&
mConnecting.compareAndSet(false, true))
{
createConnection();
mLastConnectionAttempt = System.currentTimeMillis();
if(mSocketConnector == null)
{
mSocketConnector = new NioSocketConnector();
mSocketConnector.setConnectTimeoutCheckInterval(10000);
// mSocketConnector.getFilterChain().addLast("logger",
// new LoggingFilter(IcecastTCPAudioBroadcaster.class));
mSocketConnector.getFilterChain().addLast("codec", new HttpClientCodec());
mSocketConnector.setHandler(new IcecastHTTPIOHandler());
}
mStreamingSession = null;
Runnable runnable = new Runnable()
{
@Override
public void run()
{
setBroadcastState(BroadcastState.CONNECTING);
try
{
ConnectFuture future = mSocketConnector
.connect(new InetSocketAddress(getBroadcastConfiguration().getHost(),
getBroadcastConfiguration().getPort()));
future.awaitUninterruptibly();
mStreamingSession = future.getSession();
}
catch(RuntimeIoException rie)
{
Throwable throwableCause = rie.getCause();
if(throwableCause instanceof ConnectException)
{
setBroadcastState(BroadcastState.NO_SERVER);
}
else if(throwableCause != null)
{
setBroadcastState(BroadcastState.ERROR);
mLog.debug("Failed to connect", rie);
}
else
{
setBroadcastState(BroadcastState.ERROR);
mLog.debug("Failed to connect - no exception is available");
}
disconnect();
}
}
};
getThreadPoolManager().scheduleOnce(runnable, 0l, TimeUnit.SECONDS);
}
return connected();
@@ -231,252 +171,162 @@ public class IcecastHTTPAudioBroadcaster extends AudioBroadcaster implements Pub
/**
* Disconnect from the remote server
* Disconnect from the remote broadcast server and cleanup input/output streams and socket connection
*/
public void disconnect()
{
if(mStreamingSession != null)
{
mStreamingSession.closeNow();
}
mConnecting.set(false);
}
/**
* Creates a connection to the remote server using the shoutcast configuration information. Once disconnected
* following a successful connection, attempts to reestablish a connection on a set interval
* IO Handler for managing Icecast HTTP connection and credentials
*/
private void createConnection()
public class IcecastHTTPIOHandler extends IoHandlerAdapter
{
if(mConnecting.compareAndSet(false, true))
@Override
public void sessionOpened(IoSession session) throws Exception
{
if(canConnect() && System.currentTimeMillis() - mLastConnectionAttempt >= RECONNECT_INTERVAL_MILLISECONDS)
//Send stream configuration and user credentials upon connecting to remote server
HttpRequestImpl request = new HttpRequestImpl(HttpVersion.HTTP_1_1, HttpMethod.PUT,
getConfiguration().getMountPoint(), "", getHTTPHeaders());
session.write(request);
}
private Map<String,String> getHTTPHeaders()
{
if(mHTTPHeaders == null)
{
mLog.debug("Creating a connection to icecast server");
setBroadcastState(BroadcastState.CONNECTING);
mLastConnectionAttempt = System.currentTimeMillis();
mHTTPHeaders = new HashMap<>();
mHTTPHeaders.put(IcecastHeader.ACCEPT.getValue(), "*/*");
mHTTPHeaders.put(IcecastHeader.CONTENT_TYPE.getValue(), getConfiguration().getBroadcastFormat().getValue());
mHTTPHeaders.put(IcecastHeader.USER_AGENT.getValue(), SystemProperties.getInstance().getApplicationName());
mHTTPHeaders.put(IcecastHeader.AUTHORIZATION.getValue(), getConfiguration().getBase64EncodedCredentials());
mHTTPHeaders.put(IcecastHeader.PUBLIC.getValue(), getConfiguration().isPublic() ? "1" : "0");
Uri uri = new Uri(Uri.HTTP, null, getConfiguration().getHost(), getConfiguration().getPort(),
getConfiguration().getMountPoint(), null);
BoundRequestBuilder requestBuilder = mDefaultAsyncHttpClient.preparePut(uri.toUrl());
//Use Basic (base64) authentication
Realm realm = new Realm.Builder(getConfiguration().getUserName(), getConfiguration().getPassword())
.setScheme(Realm.AuthScheme.BASIC).setUsePreemptiveAuth(true).build();
requestBuilder.setRealm(realm);
requestBuilder.setBody(this);
//We don't want this request to timeout since it will exist while we are streaming
requestBuilder.setRequestTimeout(-1);
requestBuilder.addHeader(IcecastHeader.ACCEPT.getValue(), "*/*");
requestBuilder.addHeader(IcecastHeader.CONTENT_TYPE.getValue(), getConfiguration().getBroadcastFormat().getValue());
requestBuilder.addHeader(IcecastHeader.PUBLIC.getValue(), getConfiguration().isPublic() ? "1" : "0");
StringBuilder sb = new StringBuilder();
sb.append("samplerate=").append(getConfiguration().getSampleRate()).append(";");
sb.append("quality=").append(MP3AudioConverter.AUDIO_QUALITY).append(";");
sb.append("channels=").append(getConfiguration().getChannels());
mHTTPHeaders.put(IcecastHeader.AUDIO_INFO.getValue(), sb.toString());
if(getConfiguration().hasName())
{
requestBuilder.addHeader(IcecastHeader.NAME.getValue(), getConfiguration().getName());
mHTTPHeaders.put(IcecastHeader.NAME.getValue(), getConfiguration().getName());
}
if(getConfiguration().hasDescription())
{
requestBuilder.addHeader(IcecastHeader.DESCRIPTION.getValue(), getConfiguration().getDescription());
mHTTPHeaders.put(IcecastHeader.DESCRIPTION.getValue(), getConfiguration().getDescription());
}
if(getConfiguration().hasURL())
{
requestBuilder.addHeader(IcecastHeader.URL.getValue(), getConfiguration().getURL());
mHTTPHeaders.put(IcecastHeader.URL.getValue(), getConfiguration().getURL());
}
if(getConfiguration().hasGenre())
{
requestBuilder.addHeader(IcecastHeader.GENRE.getValue(), getConfiguration().getGenre());
mHTTPHeaders.put(IcecastHeader.GENRE.getValue(), getConfiguration().getGenre());
}
if(getConfiguration().hasBitRate())
{
requestBuilder.addHeader(IcecastHeader.BITRATE.getValue(), String.valueOf(getConfiguration().getBitRate()));
mHTTPHeaders.put(IcecastHeader.BITRATE.getValue(), String.valueOf(getConfiguration().getBitRate()));
}
requestBuilder.addHeader(IcecastHeader.USER_AGENT.getValue(),
SystemProperties.getInstance().getApplicationName());
requestBuilder.addHeader(IcecastHeader.EXPECT.getValue(), HTTP_100_CONTINUE);
requestBuilder.setRequestTimeout(1000);
requestBuilder.execute(new AsyncHttpConnectionResponseHandler());
}
else
{
mConnecting.set(false);
}
return mHTTPHeaders;
}
}
public class AsyncHttpConnectionResponseHandler implements AsyncHandler<Response>
{
@Override
public void onThrowable(Throwable throwable)
public void sessionClosed(IoSession session) throws Exception
{
mLog.debug("Got a throwable!");
if(throwable instanceof ConnectException)
//If there is already an error state, don't override it. Otherwise, set state to disconnected
if(!getBroadcastState().isErrorState())
{
if(throwable.getMessage().contains("Connection refused"))
setBroadcastState(BroadcastState.DISCONNECTED);
}
mSocketConnector.dispose();
mStreamingSession = null;
mSocketConnector = null;
mConnecting.set(false);
super.sessionClosed(session);
}
@Override
public void exceptionCaught(IoSession session, Throwable throwable) throws Exception
{
if(throwable instanceof ProtocolDecoderException)
{
Throwable cause = ((ProtocolDecoderException)throwable).getCause();
if(cause instanceof HttpException &&
((HttpException)cause).getStatusCode() == HttpStatus.CLIENT_ERROR_LENGTH_REQUIRED.code())
{
setBroadcastState(BroadcastState.NO_SERVER);
//Ignore - Icecast 2.4 sometimes doesn't send any headers with their HTTP responses. Mina expects a
//content-length for HTTP responses.
}
else
{
mLog.error("HTTP protocol decoder error", throwable);
setBroadcastState(BroadcastState.TEMPORARY_BROADCAST_ERROR);
disconnect();
}
}
else
{
mLog.error("Error connecting to remote server - " + throwable.getMessage());
setBroadcastState(BroadcastState.BROADCAST_ERROR);
mLog.error("Broadcast error", throwable);
setBroadcastState(BroadcastState.TEMPORARY_BROADCAST_ERROR);
disconnect();
}
mConnecting.compareAndSet(true, false);
mConnecting.set(false);
}
@Override
public State onBodyPartReceived(HttpResponseBodyPart httpResponseBodyPart) throws Exception
public void messageReceived(IoSession session, Object object) throws Exception
{
mLog.debug("Got response body part");
return State.CONTINUE;
}
@Override
public State onStatusReceived(HttpResponseStatus httpResponseStatus) throws Exception
{
mLog.debug("Got response status: " + httpResponseStatus.toString());
State returnState = State.ABORT;
switch(httpResponseStatus.getStatusCode())
if(object instanceof DefaultHttpResponse)
{
case 200: //200-OK
setBroadcastState(BroadcastState.CONNECTED);
returnState = State.CONTINUE;
break;
case 100: //100-Continue
setBroadcastState(BroadcastState.CONNECTING);
returnState = State.CONTINUE;
break;
case 401:
setBroadcastState(BroadcastState.INVALID_PASSWORD);
break;
case 403:
String text = httpResponseStatus.getStatusText();
DefaultHttpResponse response = (DefaultHttpResponse)object;
mLog.error("Error while connecting to Icecast server: " + httpResponseStatus.getStatusCode() +
" " + text);
if(text != null)
{
if(text.contains("Content-type not supported"))
{
setBroadcastState(BroadcastState.UNSUPPORTED_AUDIO_FORMAT);
}
else if(text.contains("too many sources"))
{
setBroadcastState(BroadcastState.MAX_SOURCES_EXCEEDED);
}
else if(text.contains("Mountpoint in use"))
{
setBroadcastState(BroadcastState.MOUNT_POINT_IN_USE);
}
else
{
setBroadcastState(BroadcastState.ERROR);
}
}
else
{
switch(response.getStatus())
{
case INFORMATIONAL_CONTINUE:
break;
case SUCCESS_OK:
setBroadcastState(BroadcastState.CONNECTED);
mConnecting.set(false);
break;
case CLIENT_ERROR_UNAUTHORIZED:
setBroadcastState(BroadcastState.INVALID_CREDENTIALS);
disconnect();
break;
case CLIENT_ERROR_FORBIDDEN:
setBroadcastState(BroadcastState.CONFIGURATION_ERROR);
disconnect();
break;
default:
setBroadcastState(BroadcastState.ERROR);
}
break;
case 500:
mLog.error("Error while connecting to Icecast server: " + httpResponseStatus.getStatusCode() +
" " + httpResponseStatus.getStatusText());
setBroadcastState(BroadcastState.REMOTE_SERVER_ERROR);
break;
default:
mLog.error("Error while connecting to Icecast server - unrecognized status code: " +
httpResponseStatus.getStatusCode() + " " + httpResponseStatus.getStatusText());
setBroadcastState(BroadcastState.ERROR);
disconnect();
mLog.debug("Unspecified error: " + response.toString() + " Class:" + object.getClass());
break;
}
}
mConnecting.compareAndSet(true, false);
return returnState;
}
@Override
public State onHeadersReceived(HttpResponseHeaders httpResponseHeaders) throws Exception
{
mLog.debug("Got headers received");
return State.CONTINUE;
}
@Override
public Response onCompleted() throws Exception
{
mLog.debug("Got on completed");
setBroadcastState(BroadcastState.DISCONNECTED);
return null;
}
}
/**
* Subscription to monitor a converted audio stream subscriber and the subscriber's expressed demand for converted
* audio byte buffers
*/
public class ConvertedAudioStreamSubscription implements Subscription
{
private Subscriber<? super ByteBuffer> mSubscriber;
private AtomicLong mDemand = new AtomicLong();
/**
* Constructs a subscription
* @param subscriber to receive converted audio byte buffers
*/
public ConvertedAudioStreamSubscription(Subscriber<? super ByteBuffer> subscriber)
{
mSubscriber = subscriber;
mSubscriber.onSubscribe(this);
}
/**
* Request method for the subscriber to register demand for audio byte buffers
* @param requestedBufferCount to stream to the subscriber
*/
@Override
public void request(long requestedBufferCount)
{
mDemand.getAndAdd(requestedBufferCount);
}
/**
* Indicates if this subscription currently has registered demand from the subscriber
*/
public boolean hasDemand()
{
return mDemand.get() > 0;
}
/**
* Broadcasts the buffer to the subscriber and decrements the internal buffer demand counter
*/
public void broadcast(ByteBuffer buffer)
{
if(mSubscriber != null)
else
{
mSubscriber.onNext(buffer);
mDemand.decrementAndGet();
mLog.error("Icecast HTTP broadcaster - unrecognized message [ " + object.getClass() +
"] received:" + object.toString());
}
}
/**
* Invoked by the subscriber to cancel the stream
*/
@Override
public void cancel()
{
mLog.debug("Subscriber has requested a cancel");
stop();
}
}
}

192
src/audio/broadcast/icecast/IcecastMetadataUpdater.java
View File

@@ -18,10 +18,14 @@
******************************************************************************/
package audio.broadcast.icecast;
import audio.metadata.AudioMetadata;
import audio.metadata.Metadata;
import audio.metadata.MetadataType;
import controller.ThreadPoolManager;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.http.HttpClientCodec;
import org.apache.mina.http.HttpRequestImpl;
import org.apache.mina.http.api.HttpMethod;
@@ -36,6 +40,7 @@ import java.io.UnsupportedEncodingException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
@@ -70,14 +75,14 @@ public class IcecastMetadataUpdater
}
/**
* Initializes the network socket and HTTP headers map
* Socket connector - lazy constructor.
*/
private void init()
private NioSocketConnector getSocketConnector()
{
if (mSocketConnector == null)
{
mSocketConnector = new NioSocketConnector();
// mSocketConnector.getFilterChain().addLast("logger", new LoggingFilter(IcecastMetadataUpdater.class));
// mSocketConnector.getFilterChain().addLast("logger", new LoggingFilter(IcecastMetadataUpdater.class));
mSocketConnector.getFilterChain().addLast("http_client_codec", new HttpClientCodec());
//Each metadata update session is single-use, so we'll shut it down upon success or failure
@@ -86,22 +91,21 @@ public class IcecastMetadataUpdater
@Override
public void exceptionCaught(IoSession session, Throwable cause) throws Exception
{
// mLog.debug("Session " + session.getId() + " metadata update complete - ERROR received");
//Single-use session - close it after we receive an error
session.closeNow();
mLog.error("Metadata update failed", cause);
}
@Override
public void messageReceived(IoSession session, Object message) throws Exception
{
// mLog.debug("Session " + session.getId() + " metadata update complete - response received");
//Single-use session - close it after we receive a response
session.closeNow();
}
});
mHTTPHeaders = new HashMap<>();
mHTTPHeaders.put(IcecastHeader.USER_AGENT.getValue(), SystemProperties.getInstance().getApplicationName());
mHTTPHeaders.put(IcecastHeader.AUTHORIZATION.getValue(), mIcecastConfiguration.getBase64EncodedCredentials());
}
return mSocketConnector;
}
/**
@@ -112,91 +116,144 @@ public class IcecastMetadataUpdater
* and will remain in the update queue until the next metadata update is requested. However, this is a design
* trade-off to avoid having a scheduled runnable repeatedly processing the update queue.
*/
public void update(String metadata)
public void update(AudioMetadata audioMetadata)
{
if (metadata != null)
mMetadataQueue.offer(getSong(audioMetadata));
if (mUpdating.compareAndSet(false, true))
{
mMetadataQueue.offer(metadata);
String song = mMetadataQueue.poll();
if (mUpdating.compareAndSet(false, true))
while (song != null)
{
String metadataUpdate = mMetadataQueue.poll();
HttpRequest updateRequest = createUpdateRequest(song);
while (metadataUpdate != null)
if (updateRequest != null)
{
//Ensure we're setup for network communications (only happens once)
init();
HttpRequest updateRequest = createUpdateRequest(metadataUpdate);
if (updateRequest != null)
mThreadPoolManager.scheduleOnce(new Runnable()
{
mThreadPoolManager.scheduleOnce(new Runnable()
@Override
public void run()
{
@Override
public void run()
try
{
try
{
ConnectFuture connectFuture = mSocketConnector
.connect(new InetSocketAddress(mIcecastConfiguration.getHost(),
mIcecastConfiguration.getPort()));
connectFuture.awaitUninterruptibly();
IoSession session = connectFuture.getSession();
ConnectFuture connectFuture = getSocketConnector()
.connect(new InetSocketAddress(mIcecastConfiguration.getHost(),
mIcecastConfiguration.getPort()));
connectFuture.awaitUninterruptibly();
IoSession session = connectFuture.getSession();
if (session != null)
{
session.write(updateRequest);
}
if (session != null)
{
session.write(updateRequest);
}
catch (Exception e)
}
catch (Exception e)
{
Throwable throwableCause = e.getCause();
if (throwableCause instanceof ConnectException)
{
Throwable throwableCause = e.getCause();
if (throwableCause instanceof ConnectException)
//Do nothing, the server is unavailable
}
else
{
if (!mStackTraceLoggingSuppressed)
{
//Do nothing, the server is unavailable
}
else
{
if (!mStackTraceLoggingSuppressed)
{
mLog.error("Error sending metadata update. Future errors will " +
"be suppressed", e);
mLog.error("Error sending metadata update. Future errors will " +
"be suppressed", e);
mStackTraceLoggingSuppressed = true;
}
mStackTraceLoggingSuppressed = true;
}
}
}
}, 0l, TimeUnit.SECONDS);
}
//Fetch next metadata update to send
metadataUpdate = mMetadataQueue.poll();
}
}, 0l, TimeUnit.SECONDS);
}
mUpdating.set(false);
//Fetch next metadata update to send
song = mMetadataQueue.poll();
}
mUpdating.set(false);
}
}
/**
* Creates the song information for a metadata update
*/
private static String getSong(AudioMetadata audioMetadata)
{
StringBuilder sb = new StringBuilder();
if(audioMetadata != null)
{
Metadata to = audioMetadata.getMetadata(MetadataType.TO);
sb.append("TO:");
if(to != null)
{
if(to.hasAlias())
{
sb.append(to.getAlias().getName());
}
else
{
sb.append(to.getValue());
}
}
else
{
sb.append("UNKNOWN");
}
Metadata from = audioMetadata.getMetadata(MetadataType.FROM);
sb.append(" FROM:");
if(from != null)
{
if(from.hasAlias())
{
sb.append(from.getAlias().getName());
}
else
{
sb.append(from.getValue());
}
}
else
{
sb.append("UNKNOWN");
}
}
else
{
sb.append("Scanning ....");
}
return sb.toString();
}
/**
* Creates an HTTP GET request to update the metadata on the remote server
*/
private HttpRequest createUpdateRequest(String metadata)
private HttpRequest createUpdateRequest(String song)
{
try
{
String songEncoded = URLEncoder.encode(metadata, UTF8);
StringBuilder sb = new StringBuilder();
sb.append("mode=updinfo");
sb.append("&mount=").append(mIcecastConfiguration.getMountPoint());
sb.append("&charset=UTF%2d8");
sb.append("&song=").append(songEncoded);
sb.append("&song=").append(URLEncoder.encode(song, UTF8));
return new HttpRequestImpl(HttpVersion.HTTP_1_1, HttpMethod.GET, "/admin/metadata",
sb.toString(), mHTTPHeaders);
HttpRequestImpl request = new HttpRequestImpl(HttpVersion.HTTP_1_1, HttpMethod.GET, "/admin/metadata",
sb.toString(), getHTTPHeaders());
return request;
}
catch (UnsupportedEncodingException e)
{
@@ -206,4 +263,19 @@ public class IcecastMetadataUpdater
return null;
}
/**
* HTTP headers to use for a metadata update request.
*/
private Map<String,String> getHTTPHeaders()
{
if(mHTTPHeaders == null)
{
mHTTPHeaders = new HashMap<>();
mHTTPHeaders.put(IcecastHeader.USER_AGENT.getValue(), SystemProperties.getInstance().getApplicationName());
mHTTPHeaders.put(IcecastHeader.AUTHORIZATION.getValue(), mIcecastConfiguration.getBase64EncodedCredentials());
}
return mHTTPHeaders;
}
}

239
src/audio/broadcast/icecast/IcecastTCPAudioBroadcaster.java
View File

@@ -18,21 +18,16 @@
******************************************************************************/
package audio.broadcast.icecast;
import audio.broadcast.AudioBroadcaster;
import audio.broadcast.BroadcastState;
import audio.metadata.AudioMetadata;
import audio.metadata.Metadata;
import audio.metadata.MetadataType;
import audio.broadcast.icecast.codec.IcecastCodecFactory;
import audio.convert.MP3AudioConverter;
import controller.ThreadPoolManager;
import org.apache.mina.core.RuntimeIoException;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import properties.SystemProperties;
@@ -40,11 +35,10 @@ import properties.SystemProperties;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
public class IcecastTCPAudioBroadcaster extends IcecastAudioBroadcaster
{
private final static Logger mLog = LoggerFactory.getLogger( IcecastTCPAudioBroadcaster.class );
private final static String TERMINATOR = "\r\n";
@@ -53,10 +47,8 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
private NioSocketConnector mSocketConnector;
private IoSession mStreamingSession = null;
private IcecastMetadataUpdater mIcecastMetadataUpdater;
private long mLastConnectionAttempt = 0;
private byte[] mSilenceFrame;
private AtomicBoolean mConnecting = new AtomicBoolean();
/**
@@ -65,23 +57,16 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
*
* Note: use @see IcecastHTTPAudioBroadcaster for Icecast version 2.4.x and newer.
*
* @param configuration
* This broadcaster uses the Apache Mina library for the streaming socket connection and for metadata updates. The
* IcecastTCPIOHandler manages all interaction with the Icecast server and manages the overall broadcast state.
*
* @param configuration for the Icecast stream
*/
public IcecastTCPAudioBroadcaster(DefaultAsyncHttpClient httpClient,
ThreadPoolManager threadPoolManager,
IcecastTCPConfiguration configuration)
public IcecastTCPAudioBroadcaster(ThreadPoolManager threadPoolManager, IcecastTCPConfiguration configuration)
{
super(threadPoolManager, configuration);
}
/**
* Icecast broadcast configuration
*/
private IcecastTCPConfiguration getConfiguration()
{
return (IcecastTCPConfiguration)getBroadcastConfiguration();
}
/**
* Broadcasts the audio frame or sequence
*/
@@ -90,81 +75,13 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
{
if(connect())
{
try
if(mStreamingSession != null && mStreamingSession.isConnected())
{
send(audio);
}
catch(IOException ioe)
{
mLog.error("Error sending audio", ioe);
mStreamingSession.write(audio);
}
}
}
/**
* Broadcasts an audio metadata update
*/
@Override
protected void broadcastMetadata(AudioMetadata metadata)
{
StringBuilder sb = new StringBuilder();
if(metadata != null)
{
Metadata to = metadata.getMetadata(MetadataType.TO);
sb.append("TO:");
if(to != null)
{
if(to.hasAlias())
{
sb.append(to.getAlias().getName());
}
else
{
sb.append(to.getValue());
}
}
else
{
sb.append("UNKNOWN");
}
Metadata from = metadata.getMetadata(MetadataType.FROM);
sb.append(" FROM:");
if(from != null)
{
if(from.hasAlias())
{
sb.append(from.getAlias().getName());
}
else
{
sb.append(from.getValue());
}
}
else
{
sb.append("UNKNOWN");
}
}
else
{
sb.append("Scanning ....");
}
if(mIcecastMetadataUpdater == null)
{
mIcecastMetadataUpdater = new IcecastMetadataUpdater(getThreadPoolManager(), getConfiguration());
}
mIcecastMetadataUpdater.update(sb.toString());
}
/**
* (Re)Connects the broadcaster to the remote server if it currently is disconnected and indicates if the broadcaster
* is currently connected to the remote server following any connection attempts.
@@ -183,14 +100,14 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
if(mSocketConnector == null)
{
mLog.debug("NIO socket connector - creating");
mSocketConnector = new NioSocketConnector();
mSocketConnector.setConnectTimeoutCheckInterval(10000);
mSocketConnector.getFilterChain().addLast("logger",
new LoggingFilter(IcecastTCPAudioBroadcaster.class));
// mSocketConnector.getFilterChain().addLast("logger",
// new LoggingFilter(IcecastTCPAudioBroadcaster.class));
mSocketConnector.getFilterChain().addLast("codec",
new ProtocolCodecFilter(new TextLineCodecFactory((Charset.forName("UTF-8")))));
mLog.debug("NIO socket connector - setting handler");
new ProtocolCodecFilter(new IcecastCodecFactory()));
mSocketConnector.setHandler(new IcecastTCPIOHandler());
}
@@ -201,8 +118,6 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
@Override
public void run()
{
mLog.debug("Attempting connection to Icecast TCP Server");
setBroadcastState(BroadcastState.CONNECTING);
try
@@ -223,15 +138,16 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
}
else if(throwableCause != null)
{
setBroadcastState(BroadcastState.BROADCAST_ERROR);
setBroadcastState(BroadcastState.ERROR);
mLog.debug("Failed to connect", rie);
}
else
{
setBroadcastState(BroadcastState.ERROR);
mLog.debug("Failed to connect - no exception is available");
}
mConnecting.set(false);
disconnect();
}
}
};
@@ -249,9 +165,7 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
*/
public void disconnect()
{
mLog.info("Disconnecting Icecast TCP audio broadcaster");
if(mConnecting.get())
if(connected())
{
if(mStreamingSession != null)
{
@@ -260,77 +174,6 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
}
}
/**
* Creates an audio metadata description string that can optionally be included when connecting to the remote
* broadcast server.
*/
private String getAudioInfoMetadata()
{
StringBuilder sb = new StringBuilder();
if(getConfiguration().hasBitRate() || getConfiguration().hasChannels() || getConfiguration().hasSampleRate())
{
sb.append(IcecastHeader.AUDIO_INFO.getValue()).append(SEPARATOR);
boolean contentAdded = false;
if(getConfiguration().hasBitRate())
{
sb.append("bitrate=").append(getConfiguration().getBitRate());
contentAdded = true;
}
if(getConfiguration().hasChannels())
{
if(contentAdded)
{
sb.append(";");
}
sb.append("channels=").append(getConfiguration().getChannels());
contentAdded = true;
}
if(getConfiguration().hasSampleRate())
{
if(contentAdded)
{
sb.append(";");
}
sb.append("samplerate=").append(getConfiguration().getSampleRate());
}
}
return sb.toString();
}
/**
* Sends the string data to the remote server
*
* @param data to send
* @throws IOException if there is an error communicating with the remote server
*/
private void send(String data) throws IOException
{
if(data != null && !data.isEmpty() && mStreamingSession != null && mStreamingSession.isConnected())
{
mStreamingSession.write(data);
}
}
/**
* Sends the byte data to the remote server
* @param data to send
* @throws IOException if there is an error communicating with the remote server
*/
private void send(byte[] data) throws IOException
{
if(mStreamingSession != null && mStreamingSession.isConnected())
{
mStreamingSession.write(data);
}
}
/**
* IO Handler for managing Icecast TCP connection and credentials
*/
@@ -354,6 +197,11 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
sb.append(IcecastHeader.PUBLIC.getValue()).append(SEPARATOR)
.append(getConfiguration().isPublic() ? "1" : "0").append(TERMINATOR);
sb.append(IcecastHeader.AUDIO_INFO.getValue()).append(SEPARATOR);
sb.append("samplerate=").append(getConfiguration().getSampleRate()).append(";");
sb.append("quality=").append(MP3AudioConverter.AUDIO_QUALITY).append(";");
sb.append("channels=").append(getConfiguration().getChannels()).append(TERMINATOR);
if(getConfiguration().hasName())
{
sb.append(IcecastHeader.NAME.getValue()).append(SEPARATOR).append(getConfiguration().getName()).append(TERMINATOR);
@@ -371,19 +219,19 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
.append(getConfiguration().getDescription()).append(TERMINATOR);
}
// sb.append(getAudioInfoMetadata());
sb.append(TERMINATOR).append(TERMINATOR);
mLog.debug("Sending connection string");
session.write(sb.toString());
}
@Override
public void sessionClosed(IoSession session) throws Exception
{
setBroadcastState(BroadcastState.DISCONNECTED);
mLog.debug("Session Closed - State = Disconnected");
//If there is already an error state, don't override it. Otherwise, set state to disconnected
if(!getBroadcastState().isErrorState())
{
setBroadcastState(BroadcastState.DISCONNECTED);
}
mSocketConnector.dispose();
mStreamingSession = null;
@@ -396,10 +244,19 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
@Override
public void exceptionCaught(IoSession session, Throwable cause) throws Exception
{
mLog.debug("Exception", cause);
super.exceptionCaught(session, cause);
if(cause instanceof IOException && ((IOException)cause).getMessage().startsWith("Connection reset"))
{
mLog.info("Streaming connection reset by remote server - reestablishing connection");
disconnect();
connect();
}
else
{
mLog.error("Broadcast error", cause);
setBroadcastState(BroadcastState.ERROR);
disconnect();
}
setBroadcastState(BroadcastState.BROADCAST_ERROR);
mConnecting.set(false);
}
@@ -423,28 +280,20 @@ public class IcecastTCPAudioBroadcaster extends AudioBroadcaster
else if(message.contains("Invalid Password") ||
message.contains("Authentication Required"))
{
setBroadcastState(BroadcastState.INVALID_PASSWORD);
setBroadcastState(BroadcastState.INVALID_CREDENTIALS);
}
else
{
mLog.error("Unrecognized server response:" + message);
setBroadcastState(BroadcastState.ERROR);
}
mLog.debug("Message Received: " + message.toString());
}
}
else
{
mLog.debug("Message Received: " + object.getClass());
mLog.error("Icecast TCP broadcaster - unrecognized message [ " + object.getClass() +
"] received:" + object.toString());
}
}
@Override
public void inputClosed(IoSession session) throws Exception
{
mLog.debug("Input Closed");
super.inputClosed(session);
}
}
}

63
src/audio/broadcast/icecast/codec/IcecastCodecFactory.java Normal file
View File

@@ -0,0 +1,63 @@
/*******************************************************************************
* 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 <http://www.gnu.org/licenses/>
*
******************************************************************************/
package audio.broadcast.icecast.codec;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolEncoder;
import org.apache.mina.filter.codec.textline.LineDelimiter;
import org.apache.mina.filter.codec.textline.TextLineDecoder;
import java.nio.charset.StandardCharsets;
public class IcecastCodecFactory implements ProtocolCodecFactory
{
private ProtocolEncoder mProtocolEncoder;
private ProtocolDecoder mProtocolDecoder;
/**
* Factory for codecs to process Icecast string protocol and byte array audio frames for use with Apache Mina
*/
public IcecastCodecFactory()
{
}
@Override
public ProtocolEncoder getEncoder(IoSession ioSession) throws Exception
{
if(mProtocolEncoder == null)
{
mProtocolEncoder = new IcecastEncoder();
}
return mProtocolEncoder;
}
@Override
public ProtocolDecoder getDecoder(IoSession ioSession) throws Exception
{
if(mProtocolDecoder == null)
{
mProtocolDecoder = new TextLineDecoder(StandardCharsets.UTF_8, LineDelimiter.AUTO);
}
return mProtocolDecoder;
}
}

94
src/audio/broadcast/icecast/codec/IcecastEncoder.java Normal file
View File

@@ -0,0 +1,94 @@
/*******************************************************************************
* 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 <http://www.gnu.org/licenses/>
*
******************************************************************************/
package audio.broadcast.icecast.codec;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.AttributeKey;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolEncoderAdapter;
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
import org.apache.mina.filter.codec.textline.LineDelimiter;
import org.apache.mina.filter.codec.textline.TextLineEncoder;
import java.nio.charset.Charset;
import java.nio.charset.CharsetEncoder;
import java.nio.charset.StandardCharsets;
public class IcecastEncoder extends ProtocolEncoderAdapter
{
private static final AttributeKey ENCODER = new AttributeKey(TextLineEncoder.class, "encoder");
private static final Charset mCharset = StandardCharsets.UTF_8;
private static final int MAX_LINE_LENGTH = 2147483647;
/**
* Icecast Protocol Encoder is an Apache Mina compliant protocol encoder that supports Icecast string-based
* connection negotiation as well as byte array audio frame data.
*/
public IcecastEncoder()
{
}
@Override
public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception
{
//Audio frame data
if(message instanceof byte[])
{
byte[] data = (byte[])message;
IoBuffer buf = IoBuffer.allocate(data.length).setAutoExpand(false);
buf.put(data);
buf.flip();
out.write(buf);
}
//Icecast connection messages
else if(message instanceof String)
{
CharsetEncoder encoder = (CharsetEncoder)session.getAttribute(ENCODER);
if(encoder == null)
{
encoder = mCharset.newEncoder();
session.setAttribute(ENCODER, encoder);
}
String value = (String)message;
IoBuffer buf = IoBuffer.allocate(value.length()).setAutoExpand(true);
buf.putString(value, encoder);
if(buf.position() > MAX_LINE_LENGTH)
{
throw new IllegalArgumentException("Line length: " + buf.position());
}
else
{
buf.putString(LineDelimiter.UNIX.getValue(), encoder);
buf.flip();
out.write(buf);
}
}
else
{
throw new IllegalArgumentException("Can't encode - unrecognized object type: " + message.getClass());
}
}
}

12
src/audio/broadcast/shoutcast/v1/ShoutcastV1AudioBroadcaster.java
View File

@@ -26,7 +26,6 @@ import audio.broadcast.BroadcastMetadata;
import audio.broadcast.BroadcastState;
import audio.metadata.AudioMetadata;
import controller.ThreadPoolManager;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import record.wave.AudioPacketMonoWaveReader;
@@ -101,7 +100,7 @@ public class ShoutcastV1AudioBroadcaster extends AudioBroadcaster
catch (IOException e)
{
mLog.error("Error while dispatching audio", e);
setBroadcastState(BroadcastState.BROADCAST_ERROR);
setBroadcastState(BroadcastState.TEMPORARY_BROADCAST_ERROR);
}
}
}
@@ -261,7 +260,7 @@ public class ShoutcastV1AudioBroadcaster extends AudioBroadcaster
}
else if(response.startsWith("Invalid Password"))
{
setBroadcastState(BroadcastState.INVALID_PASSWORD);
setBroadcastState(BroadcastState.INVALID_CREDENTIALS);
return true;
}
@@ -289,7 +288,7 @@ public class ShoutcastV1AudioBroadcaster extends AudioBroadcaster
}
catch(UnknownHostException uhe)
{
setBroadcastState(BroadcastState.UNKNOWN_HOST);
setBroadcastState(BroadcastState.NO_SERVER);
mLog.error("Unknown host or port. Unable to create connection to streaming server host[" +
getShoutcastConfiguration().getHost() + "] and port[" +
getShoutcastConfiguration().getPort() + "] - will reattempt connection periodically");
@@ -320,7 +319,7 @@ public class ShoutcastV1AudioBroadcaster extends AudioBroadcaster
}
catch(UnknownHostException uhe)
{
setBroadcastState(BroadcastState.UNKNOWN_HOST);
setBroadcastState(BroadcastState.NO_SERVER);
}
catch(IOException e)
{
@@ -348,9 +347,8 @@ public class ShoutcastV1AudioBroadcaster extends AudioBroadcaster
config.setBitRate(16);
ThreadPoolManager threadPoolManager = new ThreadPoolManager();
DefaultAsyncHttpClient httpClient = new DefaultAsyncHttpClient();
final AudioBroadcaster audioBroadcaster = BroadcastFactory.getBroadcaster(httpClient, threadPoolManager,config);
final AudioBroadcaster audioBroadcaster = BroadcastFactory.getBroadcaster(threadPoolManager,config);
Path path = Paths.get("/home/denny/Music/PCM.wav");
mLog.debug("Opening: " + path.toString());

17
src/audio/broadcast/shoutcast/v2/ShoutcastV2AudioBroadcaster.java
View File

@@ -20,10 +20,9 @@ package audio.broadcast.shoutcast.v2;
import audio.AudioPacket;
import audio.broadcast.AudioBroadcaster;
import audio.broadcast.BroadcastConfiguration;
import audio.broadcast.BroadcastFactory;
import audio.broadcast.BroadcastFormat;
import audio.broadcast.BroadcastState;
import audio.broadcast.BroadcastFactory;
import audio.broadcast.shoutcast.v2.message.AuthenticateBroadcast;
import audio.broadcast.shoutcast.v2.message.ConfigureIcyGenre;
import audio.broadcast.shoutcast.v2.message.ConfigureIcyName;
@@ -41,7 +40,6 @@ import audio.broadcast.shoutcast.v2.message.UltravoxMetadata;
import audio.broadcast.shoutcast.v2.message.XMLMetadata;
import audio.metadata.AudioMetadata;
import controller.ThreadPoolManager;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import properties.SystemProperties;
@@ -77,7 +75,7 @@ public class ShoutcastV2AudioBroadcaster extends AudioBroadcaster
* Creates a Shoutcast Version 2 broadcaster.
* @param configuration details for shoutcast version 2
*/
public ShoutcastV2AudioBroadcaster(ThreadPoolManager threadPoolManager, BroadcastConfiguration configuration)
public ShoutcastV2AudioBroadcaster(ThreadPoolManager threadPoolManager, ShoutcastV2Configuration configuration)
{
super(threadPoolManager, configuration);
}
@@ -117,7 +115,7 @@ public class ShoutcastV2AudioBroadcaster extends AudioBroadcaster
catch(IOException e)
{
mLog.error("Error while dispatching audio", e);
setBroadcastState(BroadcastState.BROADCAST_ERROR);
setBroadcastState(BroadcastState.TEMPORARY_BROADCAST_ERROR);
return;
}
}
@@ -316,7 +314,7 @@ public class ShoutcastV2AudioBroadcaster extends AudioBroadcaster
if (response.isErrorResponse())
{
setBroadcastState(BroadcastState.INVALID_PASSWORD);
setBroadcastState(BroadcastState.INVALID_CREDENTIALS);
return;
}
@@ -440,7 +438,7 @@ public class ShoutcastV2AudioBroadcaster extends AudioBroadcaster
}
catch(UnknownHostException uhe)
{
setBroadcastState(BroadcastState.UNKNOWN_HOST);
setBroadcastState(BroadcastState.NO_SERVER);
mLog.error("Unknown host or port. Unable to create connection to streaming server host[" +
getShoutcastConfiguration().getHost() + "] and port[" +
getShoutcastConfiguration().getPort() + "] - will reattempt connection periodically");
@@ -471,7 +469,7 @@ public class ShoutcastV2AudioBroadcaster extends AudioBroadcaster
}
catch(UnknownHostException uhe)
{
setBroadcastState(BroadcastState.UNKNOWN_HOST);
setBroadcastState(BroadcastState.NO_SERVER);
}
catch(IOException e)
{
@@ -538,9 +536,8 @@ public class ShoutcastV2AudioBroadcaster extends AudioBroadcaster
else
{
ThreadPoolManager threadPoolManager = new ThreadPoolManager();
DefaultAsyncHttpClient httpClient = new DefaultAsyncHttpClient();
final AudioBroadcaster audioBroadcaster = BroadcastFactory.getBroadcaster(httpClient, threadPoolManager,config);
final AudioBroadcaster audioBroadcaster = BroadcastFactory.getBroadcaster(threadPoolManager,config);
Path path = Paths.get("/home/denny/Music/PCM.wav");
mLog.debug("Opening: " + path.toString());

24
src/audio/convert/MP3AudioConverter.java
View File

@@ -33,7 +33,7 @@ import java.util.List;
public class MP3AudioConverter implements IAudioConverter
{
private final static Logger mLog = LoggerFactory.getLogger( MP3AudioConverter.class );
public static final int AUDIO_QUALITY = Lame.QUALITY_LOW;
private LameEncoder mEncoder;
private ByteArrayOutputStream mMP3Stream = new ByteArrayOutputStream();
private byte[] mMP3Buffer;
@@ -47,7 +47,7 @@ public class MP3AudioConverter implements IAudioConverter
public MP3AudioConverter(int bitRate, boolean variableBitRate)
{
mEncoder = new LameEncoder(AudioFormats.PCM_SIGNED_8KHZ_16BITS_MONO,
bitRate, MPEGMode.MONO, Lame.QUALITY_LOW, variableBitRate);
bitRate, MPEGMode.MONO, AUDIO_QUALITY, variableBitRate);
mMP3Buffer = new byte[mEncoder.getPCMBufferSize()];
}
@@ -65,13 +65,21 @@ public class MP3AudioConverter implements IAudioConverter
int pcmBytesPosition = 0;
while (0 < (mp3BufferSize = mEncoder.encodeBuffer(pcmBytes, pcmBytesPosition, pcmBufferSize, mMP3Buffer)))
try
{
pcmBytesPosition += pcmBufferSize;
pcmBufferSize = Math.min(mMP3Buffer.length, pcmBytes.length - pcmBytesPosition);
mMP3Stream.write(mMP3Buffer, 0, mp3BufferSize);
}
while (0 < (mp3BufferSize = mEncoder.encodeBuffer(pcmBytes, pcmBytesPosition, pcmBufferSize, mMP3Buffer)))
{
pcmBytesPosition += pcmBufferSize;
pcmBufferSize = Math.min(mMP3Buffer.length, pcmBytes.length - pcmBytesPosition);
mMP3Stream.write(mMP3Buffer, 0, mp3BufferSize);
}
return mMP3Stream.toByteArray();
return mMP3Stream.toByteArray();
}
catch(Exception e)
{
mLog.error("There was an error converting audio to MP3: " + e.getMessage());
return new byte[0];
}
}
}

1
src/module/decode/p25/P25DecoderEditor.java
View File

@@ -137,7 +137,6 @@ public class P25DecoderEditor extends ValidatingEditor<Channel>
@Override
public void validate( Editor<Channel> editor ) throws EditorValidationException
{
mLog.debug( "Validating editor:" + editor.getClass() );
/**
* Enforce CQPSK modulation uses a Tuner Source for I/Q sample data
*/