voicespeakeroutput.java

Myjxta的源代码 基于JXTA的P2P即时通信系统

/*
 * Created on Oct 17, 2004
 *
 * TODO To change the template for this generated file go to
 * Window - Preferences - Java - Code Style - Code Templates
 */
package net.jxta.myjxta.plugins.vojxta;

import net.jxta.endpoint.ByteArrayMessageElement;
import net.jxta.endpoint.MessageElement;
import net.jxta.logging.Logging;
import net.jxta.myjxta.dialog.DialogListener;
import net.jxta.myjxta.dialog.DialogMessage;

import javax.sound.sampled.*;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author Ravi
 * @author jamoore
 * @modified 2005-03-05 jamoore make this the dialog listener for all incoming
 * @modified 2005-03-06 jamoore remove voiceplayer interface
 * @modified 2005-03-06 jamoore rework threads to spin up only on buffer additions
 * @modified 2005-03-05 jamoore add gain control accessors
 * @modified 2005-03-07 jamoore add statistical accessors, vars
 * @modified 2005-03-18 jamoore refactor
 * @modified 2005-03-26 jamoore add message ack thread
 * @modified 2005-03-28 jamoore add voice message timeout thread
 * @modified 2005-03-30 jamoore add thread to play decoded pcm in block sizes
 * @modified 2005-03-31 jamoore let audio system handle source line buffer size
 * @modified 2005-04-01 jamoore sourceline writes are now pcmBuffer.size()
 */
public final class VoiceSpeakerOutput extends Thread implements DialogListener, AudioResource {

    private static final Logger LOG = Logger.getLogger(VoiceSpeakerOutput.class.getName());

    /**
     * speaker line out
     */
    private SourceDataLine sourceLine = null;
    ;

    /**
     * wrapper form speex decoder
     */
    private Decoder decoder = null;
    ;

    /**
     * static values for now till first testing phase is over.
     * represents the number of blocks that a buffer will hold
     */
    private static final int DEFAULT_BUFFER_SIZE_MULTIPLIER = 50;

    /**
     * static values for now till first testing phase is over.
     * represents the number of blocks in a message that a buffer will hold
     * increasing this value will reduce the number of messages and possibly
     * increase the latency of audio to the endpoint
     */
    private static final int DEFAULT_MESSAGE_SIZE_MULTIPLIER = 10;

    /**
     * denots state for this output line
     * ON := line started (line must be in open state)
     * PAUSE := line stoped (line must be opend and started)
     * OFF := line stopped and closed
     */

    private static final int STATE_ON = 1;

    private static final int STATE_OFF = 0;

    private static final int STATE_PAUSE = 2;

    /**
     * denotes the current speaker state
     */
    private int speakerState = STATE_PAUSE;

    /**
     * command that controls this hardwawre resource
     */
    private VoJxtaCallControl callControl = null;

    /**
     * incomming message buffer, these messages contain encoded speex bytes
     */
    private VoiceDataBuffer recvBuffer = null;

    /**
     * statistic
     */
    private long receivedMessages = 0;

    /**
     * statistic
     */
    private long receivedBytes = 0;

    //dubugging .. remove when finished
    final boolean firstReceivedMessage = true;
    final byte[] firstReceivedBytes = null;

    /**
     * wait lock - pauses data flow to this hardware resource (speaker)
     */
    private Object pauseLock = null;

    /**
     * wait lock - waits on received messages to fill buffer sufficiently
     */
    private Object bufferLock = null;

    /**
     * wait lock - waits on pcm blocks to fill the  sourceLine buffer.
     */
    private Object sourceLineThreadLock = null;

    /**
     * true if we are holding on the sourceLinebuffer to fill
     */
    private Boolean sourceLineThreadWaiting = null;

    /**
     * true if we are holding on the encoded message buffer to fill up
     */
    private Boolean messageThreadWaiting = null;

    /**
     * basic block size of raw audio data that speex encode operates on. On
     * decode this is the block size of the decoded speex bytes
     */
    private final int rawChunkSize = AudioResource.BLOCK_SIZE;


    /**
     * basic block size of incoming encoded speex bytes. this value is
     * dependent on the quality level of speex encoding.
     */
    private int speexChunkSize = 0;
    ;


    /**
     * represents the number of speex chunks to pack into one message. this
     * value can affect latency if too large. should never be larger than
     * speexBufferSize
     */
    private int speexMessageSize = 0;

    /**
     * donotes the size in bytes of the lines (speakers) buffer
     */
    private final int rawBufferSize = rawChunkSize * 60;

    /**
     * represents the size of the buffer holding incoming speex  bytes
     */
    private int speexBufferSize = 0;

    /**
     * donotes the default mixer obtained from the line (speaker). we obtain
     * controls such as volume, mute from this object
     */
    private Mixer speakerMixer = null;

    /**
     * denotes the gain control on this line
     */
    private FloatControl gainControl = null;

    /**
     * denotes the mute control on this line
     */
    private BooleanControl muteControl = null;

    /**
     * statistic
     */
    private long averageDecodeTime = 0;

    /**
     * Human readable peer endpoint name
     */
    private String originator = null;

    /**
     * System time in milliseconds last voice message came in
     */
    private long timeOfLastVoiceMessage = 0;

    /**
     * controls writing to the sourceLine (speaker). Writing small blocks
     * (ie 640b) will produce backlogs and latency.  This thread scopes the
     * complexity of dealing with the sourceLine buffer and performs the actual
     * writing to sourceLine.
     */
    private WriteToSourceLineThread sourceLineThread = null;

    /**
     */
    public VoiceSpeakerOutput(VoJxtaCallControl callControl) {

        LOG.setLevel(Level.INFO);
        //LOG.setLevel (Level.SEVERE);

        this.setPriority(Thread.NORM_PRIORITY);

        pauseLock = new Object();

        bufferLock = new Object();

        sourceLineThreadLock = new Object();

        this.callControl = callControl;

        getCallControl().getDialog().addListener(this);

        messageThreadWaiting = new Boolean(false);

        sourceLineThreadWaiting = new Boolean(false);

        /** prime the pump*/
        timeOfLastVoiceMessage = System.currentTimeMillis();

        //printMixers(); printControls();

        //printSupportedControls();
    }

    /**
     * Seeks out a sourceline (speaker) of the specified format. Opens that line
     * and finds its mute and gain controls.
     */
    public void obtainHardware() {

        AudioFormat format = new AudioFormat(AudioFormat.Encoding.PCM_SIGNED, 16000, 16, 1, 2, 16000, false);

        DataLine.Info info = new DataLine.Info(SourceDataLine.class, format, AudioSystem.NOT_SPECIFIED);//rawBufferSize);

        try {

            sourceLine = (SourceDataLine) AudioSystem.getLine(info);

        } catch (LineUnavailableException e) {

            e.printStackTrace();

        }

        try {

            sourceLine.open();

        } catch (LineUnavailableException e) {

            e.printStackTrace();

        }

        if (sourceLine.isControlSupported(FloatControl.Type.MASTER_GAIN)) {

            gainControl = (FloatControl) sourceLine.getControl(FloatControl.Type.MASTER_GAIN);

        }

        if (sourceLine.isControlSupported(BooleanControl.Type.MUTE)) {

            muteControl = (BooleanControl) sourceLine.getControl(BooleanControl.Type.MUTE);
        }

    }

    /**
     * Starts source line (speaker). Set thread state to Start. Initialize
     * incoming buffer. start sourceline thread.
     */
    public void beginSpeaker() {

        decoder = new Decoder();

        speexChunkSize = ((Integer) VoJxtaCallControl.qualityByteMap.get(new Integer(getCallControl().getMinimumVoiceQuality()))).intValue();
        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("Speaker chunk size " + speexChunkSize);
        }
        if (speexMessageSize == 0) {

            speexMessageSize = speexChunkSize * DEFAULT_MESSAGE_SIZE_MULTIPLIER;

        }
        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("Speaker message size " + speexMessageSize);
        }
        if (speexBufferSize == 0) {

            speexBufferSize = speexChunkSize * DEFAULT_BUFFER_SIZE_MULTIPLIER;
        }

        if (this.recvBuffer == null) {

            this.recvBuffer = new VoiceDataBuffer(speexBufferSize, this, "SpeakerControl");
        }


        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("Speaker buffer size " + speexBufferSize);
        }

        if (getSpeakerState() == STATE_PAUSE) {

            synchronized (pauseLock) {

                pauseLock.notify();
            }

            sourceLine.start();

            if (sourceLine.isControlSupported(FloatControl.Type.MASTER_GAIN)) {

                gainControl = (FloatControl) sourceLine.getControl(FloatControl.Type.MASTER_GAIN);

            }

            if (sourceLine.isControlSupported(FloatControl.Type.MASTER_GAIN)) {

                muteControl = (BooleanControl) sourceLine.getControl(BooleanControl.Type.MUTE);

            }
        }

        setSpeakerState(this.STATE_ON);

        sourceLineThread = new WriteToSourceLineThread();

        sourceLineThread.start();

        super.start();
    }

    /**
     * Stops writting to line. Race condition exists. Fix
     */
    public void endSpeaker() {

        setSpeakerState(this.STATE_OFF);
        if (sourceLine != null) {
            sourceLine.stop();
        }

    }

    /**
     * Gives speaker control back to the system. Null's resources.
     */
    public void releaseHardware() {

        gainControl = null;

        muteControl = null;

        if (sourceLine != null) {
            sourceLine.flush();

            sourceLine.drain();

            sourceLine.stop();

            sourceLine.close();
        }


    }


    /**
     * Interface from dialogListener. Receives all incomming messages on this
     * pipe. sorts and diseminates.
     */
    public void receive(DialogMessage msg) {

        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            //LOG.info ("Begin receive (DialogMessage)");
        }
        if (this.originator == null) {

            setOriginator(msg.getOriginator());
        }
        /** retrieve the messages command */
        String sessionCommand = getMessageSessionCommand(msg);

        /** vojxta data is processed and sent to audio out
         *   this will be the dominant command by far so lets get this out
         *   immediately */
        if (sessionCommand.equals(VoJxtaCallControl.COMMAND_VOJXTA_DATA)) {

            if (getCallControl().getProtocolState() == VoJxtaCallControl.SESSION_VOJXTA_INCALL) {

                receiveVoJxtaData(msg);
            }

        } else {
            /** otherwise it is a session command that we need to deal with */
            getCallControl().callControl(sessionCommand, msg);

        }
    }


    /**
     * Sets the human readable name of the remote peer
     */
    protected void setOriginator(String originator) {

        this.originator = originator;
    }

    /**
     * Gets the human readable name of the remote peer
     */
    public String getOriginator() {

        return this.originator;
    }

    /**
     * Returns a long representing the system time in milliseconds a voice data
     * message was received from remote host.
     */
    public long getTimeOfLastVoiceMessage() {

        return this.timeOfLastVoiceMessage;
    }

    /**
     * Deconstructs the voice data message. If we are waiting on new data to
     * write to source line (speaker) out then check the buffer size for
     * sufficient bytes notify play thread.
     */
    public void receiveVoJxtaData(DialogMessage msg) {

        this.timeOfLastVoiceMessage = System.currentTimeMillis();

        byte[] voiceData = getMessageVoiceBytes(msg);


        if (voiceData != null) {

            recvBuffer.append(voiceData);

            this.receivedMessages += 1;

            this.receivedBytes += voiceData.length;

            boolean waiting = false;

            synchronized (messageThreadWaiting) {

                waiting = messageThreadWaiting.booleanValue();
            }
            if (waiting && recvBuffer.size() >= speexChunkSize) {

                messageThreadWaiting = new Boolean(false);

                synchronized (bufferLock) {

                    bufferLock.notify();
                }
            }

        } else {
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("reveiceVoJxtaData : voice data is null");
            }
        }

    }


    /**
     * Decode speex block, write to line out. Pause lock goes active on buffer
     * starvation. speex chunk size is wholey dependent on speex encode quality.
     * Quality is determined by message excahnge in VojxtaCallControl. The lesser
     * quality wins.
     */
    public void run() {


        while (true) {

            try {

                if (getSpeakerState() == this.STATE_OFF) {

                    if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                        LOG.info("run : is off");
                    }
                    //we should send any remaining data in buffer first
                    break;
                }

                if (recvBuffer.size() >= (speexMessageSize)) {

                    decodeAndPlay(recvBuffer.get(speexMessageSize));

                } else {
                    synchronized (messageThreadWaiting) {

                        messageThreadWaiting = new Boolean(true);
                    }

                    synchronized (bufferLock) {
                        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                            //LOG.info ("Waiting for new messages..");
                        }

                        bufferLock.wait();
                    }
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        }


    }

    /**
     * Returns the current size in bytes of data stored in the receive buffer
     */
    public int getBufferSize() {

        return recvBuffer.size();
    }

    /**
     * Returns the size in bytes of the capacity of the receive buffer
     */
    public int getBufferCapacity() {

        return recvBuffer.getCapacity();
    }


    /**
     * Return out local state, ON OFF PAUSE
     */
    private void setSpeakerState(int speakerState) {

        this.speakerState = speakerState;

    }

    /**
     * Return the call control object
     */
    protected VoJxtaCallControl getCallControl() {

        return this.callControl;
    }