android 声音流媒体rtsp解决方案

package cn.bviews.android.RTSPStreamAudioPlayer; import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.NetworkInterface; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import java.util.Enumeration; import java.util.Iterator; import java.util.concurrent.atomic.AtomicBoolean; import android.util.Log; public class RTSPConnect extends Thread implements IEvent { private static final String TAG = "RTSPConnecti"; private static final String RTSP_VERSION = " RTSP/1.0\r\n"; private static final String RTSP_OK = "RTSP/1.0 200 OK"; /* 远程主机地址 */ private final InetSocketAddress remoteAddress; /* 本机地址 */ private final InetSocketAddress localAddress; /* 数据源地址 */ private String address; /* 发送缓冲区 */ private final ByteBuffer sendBuf; /* 接收缓冲区 */ private final ByteBuffer receiveBuf; private static final int BUFFER_SIZE = 10240; /* 端口选择器 */ private Selector selector; private SocketChannel socketChannel; private RTPReceiveUDP rtp; /* RTP 接收端口 */ private int RTPPort; /* 连接状态 */ private Status sysStatus; private String sessionId; private String trackInfo; private int seq = 1; private boolean isSended; private AtomicBoolean shutdown; private enum Status { init, options, describe, getparameter, setup, play, setparameter, pause, teardown } private String getLocalAddress() { Enumeration<NetworkInterface> netInterfaces = null; StringBuffer sb = new StringBuffer(""); try { netInterfaces = NetworkInterface.getNetworkInterfaces(); while (netInterfaces.hasMoreElements()) { NetworkInterface ni = netInterfaces.nextElement(); Enumeration<InetAddress> ips = ni.getInetAddresses(); while (ips.hasMoreElements()) { if(sb.toString().length() > 0) return sb.toString();; InetAddress local = ips.nextElement(); sb.append(local.getHostAddress()); } } } catch (Exception e) { e.printStackTrace(); } return sb.toString(); } public RTSPConnect(InetSocketAddress remoteAddress, InetSocketAddress localAddress, String address) { this.remoteAddress = remoteAddress; this.localAddress = localAddress; this.address = address; //初始化缓冲区 sendBuf = ByteBuffer.allocateDirect(BUFFER_SIZE); receiveBuf = ByteBuffer.allocateDirect(BUFFER_SIZE); if(selector == null) { try{ selector = Selector.open(); }catch(Exception e) { Log.e(TAG, "Selector open ERROR"); e.printStackTrace(); } } startup(); sysStatus = Status.init; shutdown = new AtomicBoolean(false); isSended = false; } public RTSPConnect(InetSocketAddress remoteAddress, String address) { // TODO Auto-generated constructor stub this.remoteAddress = remoteAddress; this.localAddress = new InetSocketAddress(getLocalAddress(), 0); this.address = address; //初始化缓冲区 sendBuf = ByteBuffer.allocateDirect(BUFFER_SIZE); receiveBuf = ByteBuffer.allocateDirect(BUFFER_SIZE); if(selector == null) { try{ selector = Selector.open(); }catch(Exception e) { Log.e(TAG, "Selector open ERROR"); e.printStackTrace(); } } startup(); sysStatus = Status.init; shutdown = new AtomicBoolean(false); isSended = false; } public void startup() { try{ socketChannel = SocketChannel.open(); //打开端口 socketChannel.configureBlocking(false); socketChannel.socket().bind(localAddress); Log.e(TAG, "isBound:" + socketChannel.socket().isBound() +" Port:"+socketChannel.socket().getLocalPort()); if(socketChannel.connect(remoteAddress)) { Log.i(TAG, "socket Channel start to Connect " + remoteAddress); } socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ | SelectionKey.OP_WRITE, this); Log.i(TAG, " port open Success"); }catch(Exception e) { e.printStackTrace(); } rtp = new RTPReceiveUDP(); } /* 判断是否连接 */ private boolean isConnected() { return socketChannel != null && socketChannel.isConnected(); } private byte[] recieve() { // TODO Auto-generated method stub if(isConnected()) { try{ int len = 0; int readBytes = 0; synchronized (receiveBuf) { receiveBuf.clear(); try{ while((len = socketChannel.read(receiveBuf)) > 0) { readBytes += len; } }finally{ receiveBuf.flip(); } if(readBytes > 0) { final byte[] tmp = new byte[readBytes]; receiveBuf.get(tmp); return tmp; }else { Log.i(TAG, "接收数据为空"); return null; } } }catch(final IOException e) { Log.e(TAG, "接收消息为空"); e.printStackTrace(); } }else{ Log.e(TAG, "端口没有连接"); } return null; } private void handle(byte[] msg) { // TODO Auto-generated method stub String tmp = new String(msg); Log.i(TAG, tmp); if(tmp.startsWith(RTSP_OK)) { switch(sysStatus) { case init: sysStatus = Status.options; break; case options: sysStatus = Status.describe; trackInfo = "Audio" ; //默认为Audio, break; case describe: sysStatus = Status.getparameter; break; case getparameter: sessionId = tmp.substring(tmp.indexOf("Session: ") + 9, tmp.indexOf(";timeout=")); Log.e(TAG, "sessionID:" + sessionId); if(sessionId != null && sessionId.length() > 0) { sysStatus = Status.setup; } break; case setup: sysStatus = Status.play; break; case play: sysStatus = Status.setparameter; break; case setparameter: sysStatus = Status.pause; break; case pause: sysStatus = Status.teardown; break; case teardown: sysStatus = Status.init; break; default: Log.e(TAG, "MSG ERROR: " + tmp); break; } isSended = false; }else { Log.e(TAG, "RETURN ERROR: " + tmp); } } //@Override public void connect(SelectionKey key) throws IOException { // TODO Auto-generated method stub if(isConnected()) { return; } socketChannel.finishConnect(); while(!socketChannel.isConnected()) { try { Thread.sleep(300); }catch(final InterruptedException e) { e.printStackTrace(); } socketChannel.finishConnect(); } } //@Override public void error(Exception e) { // TODO Auto-generated method stub e.printStackTrace(); } //@Override public void read(SelectionKey key) throws IOException { // TODO Auto-generated method stub final byte[] msg = recieve(); if(msg != null) { handle(msg); }else { key.cancel(); } } //@Override public void write() throws IOException { // TODO Auto-generated method stub if(isConnected()) { try{ socketChannel.write(sendBuf); }catch(final IOException e) {e.printStackTrace();} }else { Log.e(TAG, "write ERROR:通道为空或者未连接"); } } @Override public void run() { while(!shutdown.get()) { try{ if (isConnected() && (!isSended)) { switch(sysStatus){ case init: doOption(); break; case options: doDescribe(); break; case describe: doGetParameter(); break; case getparameter: rtp.start(); doSetup(); break; case setup: if(sessionId == null && sessionId.length() > 0) { Log.e(TAG, "setup ERROR");