rtpstream.cpp

语音接口～语音识别 & 从声音识别到对各种情报的查找/提供 资料请求～住所.姓名.电话号码等的识别接待 受订货业务～被定型化的受订货业务 预约业务～预约情况的向导和预约接

// Copyright (C) 2005 Open Source Telecom Corp.
//  
// 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 2 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, write to the Free Software 
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.

#include "driver.h"

namespace sipdriver {
using namespace ost;
using namespace std;

RTPStream::RTPStream(Session *s) :
SymmetricRTPSession(s->getLocalAddress(), s->getLocalPort()), 
AudioBase(&s->info), 
TimerPort(), Mutex()
{
        session = s;
        dropInbound = false;
        oTimestamp = 0;
	source = NULL;
	sink = NULL;
	tone = NULL;
	lastevt = 256;
	dtmfcount = 0;
	stopid = MSGPORT_WAKEUP;
	dtmf = NULL;
	fcount = 0;
	silent_frame = new Sample[info.framecount];
	memset(silent_frame, 0, info.framecount * 2);
	silent_encoded = new unsigned char[info.framesize];
        if(info.encoding == Audio::pcm16Mono)
	{
		info.order = __BIG_ENDIAN;
                memset(silent_encoded, 0, info.framesize);
	}
        else if(session->codec)
                session->codec->encode(silent_frame, silent_encoded, info.framecount);

	pbuffer = new unsigned char[info.framesize];
	buffer = new unsigned char[info.framesize];
	lbuffer = new Sample[info.framecount];
}

RTPStream::~RTPStream()
{
	terminate();
	delete[] silent_frame;
	delete[] silent_encoded;
	delete[] lbuffer;
	delete[] buffer;
	delete[] pbuffer;
	if(dtmf)
	{
		delete dtmf;
		dtmf = NULL;
	}
}

void RTPStream::start(void)
{
	DynamicPayloadFormat pf(session->data_payload, info.rate);
	
	setPayloadFormat(pf);	
	setMaxSendSegmentSize(1024);

	source = NULL;
	sink = NULL;
	tonecount = 0;

        SymmetricRTPSession::startRunning();
}

void RTPStream::set2833(struct dtmf2833 *data, timeout_t duration)
{
	enter();
	jsend = Driver::sip.jitter;
	oTimestamp = 0;
	dtmfcount = (duration + (info.framing - 1)) / info.framing;
	memcpy(&dtmfpacket, data, sizeof(dtmfpacket));
	dtmfpacket.ebit = 0;
	dtmfpacket.rbit = 0;
	dtmfpacket.vol = 12;
	dtmfpacket.duration = 0;
	leave();
}	

void RTPStream::setTone(AudioTone *t, timeout_t max)
{
	enter();
	jsend = Driver::sip.jitter;
	tone = t;
	fcount = max / info.framing;
	oTimestamp = 0;
	leave();
}

void RTPStream::setSource(AudioBase *get, timeout_t max)
{
        enter();
	jsend = Driver::sip.jitter + 1;
	ending = false;
        source = get;
	fcount = max / info.framing;
	oTimestamp = 0;
        leave();
}

void RTPStream::setSink(AudioBase *put, timeout_t max)
{
        enter();
        sink = put;
	fcount = max / info.framing;
        leave();
}

bool RTPStream::onRTPPacketRecv(IncomingRTPPkt &pkt)
{
	Event event;
	struct dtmf2833 *evt;
	unsigned pid = session->dtmf_payload;
	AudioCodec *codec;
	Level level;

        // this is where we test for 2833!!!

	if(!pid || pid != pkt.getPayloadType())
		goto check2;

	session->dtmf_inband = false;
	evt = (struct dtmf2833 *)pkt.getPayload();

	if(evt->event != lastevt)
		goto post;

	if(tonecount)
		goto checkbit;

	if(evt->ebit)
		goto checkbit;

post:
	lastevt = evt->event;

	memset(&event, 0, sizeof(event));
	switch(evt->event)
	{
        case 0:
        case 1:
        case 2:
        case 3:
        case 4:
        case 5:
        case 6:
        case 7:
        case 8:
        case 9:
        case 10:
        case 11:
        case 12:
        case 13:
        case 14:
        case 15:
		event.id = DTMF_KEYUP;
		event.dtmf.digit = evt->event;
		event.dtmf.duration = 60;
		break;
	case 16:
		event.id = LINE_PICKUP;
		break;
	case 64:
		event.id = LINE_OFF_HOOK;
		break;
	case 65:
		event.id = LINE_ON_HOOK;
		break;
	case 66:
	case 67:
	case 68:
	case 69:
		event.id = TONE_START;
		event.tone.name = "dialtone";
		break;
	case 70:
	case 71:
		event.id = TONE_START;
		event.tone.name = "ringback";
		break;
	case 72:
		event.id = TONE_START;
		event.tone.name = "busytone";
		break;
	case 73:
		event.id = TONE_START;
		event.tone.name = "reorder";
		break;
	case 74:
		event.id = TONE_START;
		event.tone.name = "intercept";
		break;
	case 76:
		event.id = CALL_HOLD;
		break;
	case 78:
	case 79:
		event.id = TONE_START;
		event.tone.name = "waiting";
		break;
	case 160:
	case 161:
		event.id = LINE_WINK;
		break;
	default:
		return false;
	}				

	session->queEvent(&event);

checkbit:
	if(evt->ebit && tonecount && stopid != MSGPORT_WAKEUP)
	{
		memset(&event, 0, sizeof(event));
		event.id = stopid;
		stopid = MSGPORT_WAKEUP;
		session->queEvent(&event);
		tonecount = 0;
	}
	else if(evt->ebit)
		tonecount = 0;
	else
		tonecount = 3;
	return false;

check2:
	lastevt = 256;

	if(pkt.getPayloadType() != session->data_payload)
		return false;

	if(sink || session->peer)
		goto silence;

        if(session->dtmf_inband)
		goto silence;

	return false;

silence:
	level = Driver::sip.silence;
	codec = session->codec;
	if(!level || !codec)
		return true;

	// bug...
	if(session->info.encoding == gsmVoice)
		return true;

	if(codec->isSilent(level, (void *)pkt.getRawPacket(), session->info.framecount))
		return false;

	return true;
}

void RTPStream::postAudio(Encoded encoded)
{
	register BayonneSession *peer;

	peer = session->peer;
	if(!peer || !session->isJoined())
		return;

	if(peer->peerLinear())
	{
		session->codec->decode(lbuffer, encoded, info.framecount);
		peer->peerAudio((Encoded)lbuffer);
	}
	else
		peer->peerAudio(encoded);
}

void RTPStream::peerAudio(Encoded encoded, bool linear)
{
	jsend = Driver::sip.jitter;

	if(tone || dtmfcount)
		return;

	if(linear && session->codec)
	{
		session->codec->encode((Linear)encoded, pbuffer, info.framecount);
		putNative(pbuffer, info.framesize);
	}
	else
		putNative(encoded, info.framesize);
}

void RTPStream::run(void)
{
        size_t r, n;
        size_t size = info.framesize;
        timeout_t timer;
	bool first = true;
	Event event;
	unsigned count;
	AudioCodec *codec = session->codec;
	bool dtmfdigits = false;
	char dbuf[65];
	int dig;
	Linear tbuf;
	bool silent;
	DynamicPayloadFormat pf_data(session->data_payload, info.rate);
	DynamicPayloadFormat pf_dtmf(session->dtmf_payload, info.rate);
	unsigned short pps = (unsigned short)(1000 / info.framing);
	unsigned short dtmfoffset = (unsigned short)(info.rate / pps);

	jitter = Driver::sip.jitter;
	jsend = Driver::sip.jitter;

	if(session->dtmf_inband)
		dtmf = new DTMFDetect();

        setTimer(0);

	setCancel(cancelDeferred);
        for(;;)
        {
		silent = true;
		if(dtmfdigits)
		{
			dtmf->getResult(dbuf, 64);
			dig = -1;
			if(dbuf[0])
				dig = Bayonne::getDigit(dbuf[0]);
			if(dig > -1)
			{
				memset(&event, 0, sizeof(event));
				event.id = DTMF_KEYUP;
				event.dtmf.digit = dig;
				event.dtmf.duration = 60;
				session->queEvent(&event);
			}			
			dtmfdigits = false;
		}
                incTimer(info.framing);
                enter();

		// inband detector disabled if dtmf supported elsewhere

		if(dtmf && !session->dtmf_inband)
		{
			delete dtmf;
			dtmf = NULL;
			dtmfdigits = false;
		}

		if(dtmfcount)
		{
			--dtmfcount;
			if(!dtmfcount)
				dtmfpacket.ebit = 1;
			setPayloadFormat(pf_dtmf);
			put2833(&dtmfpacket);
			pps = ntohs(dtmfpacket.duration) + dtmfoffset;
			dtmfpacket.duration = htons(pps);
			if(!dtmfcount)
				setPayloadFormat(pf_data);
			if(jsend)
			{
				--jsend;
				leave();
				dispatchDataPacket();
				continue;
			}
		}
		else if(tone)
		{
			tbuf = tone->getFrame();
			if(!tbuf || !codec)
			{
				fcount = 0;
				tone = NULL;
				memset(&event, 0, sizeof(event));
				event.id = AUDIO_IDLE;
				session->queEvent(&event);
				goto done;
			}
			if(codec)
			{
				codec->encode(tbuf, buffer, info.framecount);
				putNative(buffer, size);
				silent = false;
			}				
			if(jsend)
			{
				--jsend;
				leave();
				dispatchDataPacket();
				continue;
			}
			
		}
                else if(source)
                {
                        n = source->getBuffer(buffer, size);
			if(n == size)
			{
				if(ending)
				{
					memset(&event, 0, sizeof(event));
					event.id = AUDIO_ACTIVE;
					session->queEvent(&event);
					ending = false;
				}
				silent = false;
	                        putNative(buffer, size);
			}
			else if(!ending)
			{
				fcount = 0;
				memset(&event, 0, sizeof(event));
				event.id = AUDIO_IDLE;
				session->queEvent(&event);
				ending = true;
			}
			if(jsend && !silent)
			{
				--jsend;
				leave();
				dispatchDataPacket();
				continue;
			}
                }
		else if(!session->peer)
			jsend = Driver::sip.jitter;

done:
		leave();
		if(silent && Driver::sip.data_filler)
		{
			silent = false;
			putNative(silent_encoded, info.framesize);
		}

		r = dispatchDataPacket();

		if(tonecount == 1 && stopid != MSGPORT_WAKEUP)
		{
			memset(&event, 0, sizeof(event));
			event.id = stopid;
			stopid = MSGPORT_WAKEUP;
			session->postEvent(&event);
		}			

		if(tonecount)
			--tonecount;

		if(first)
		{
			while(isPendingData(0))
			{
        	                r = takeInDataPacket();
                	        if(r < 0)
                        	        break;

				Thread::yield();
				jitter = Driver::sip.jitter;

				if(first)
				{
					memset(&event, 0, sizeof(event));
					event.id = AUDIO_START;
					session->queEvent(&event);
					count = 6;
					while(dtmf && count-- && session->dtmf)
						dtmf->putSamples(silent_frame, info.framecount);
				}

				first = false;

	                        enter();
        	                n = getNative(buffer, r);
				if(dtmf && codec && n == size && session->dtmf)
				{
					codec->decode(lbuffer, buffer, info.framecount);
					Thread::yield();
					dtmfdigits = dtmf->putSamples(lbuffer, info.framecount) > 0;					
				}
                	        if(sink)
                        	        sink->putBuffer(buffer, n);
				else
					postAudio(buffer);
                        	leave();
				Thread::yield();
	                }
		}
		else if(!jitter)
		{
			if(isPendingData(0))
			{
				r = takeInDataPacket();
				if(r < 0)
					break;

				Thread::yield();
				enter();
				n = getNative(buffer, r);
				if(dtmf && codec && n == size && session->dtmf)
				{
					codec->decode(lbuffer, buffer, info.framecount);
					Thread::yield();
					dtmfdigits = dtmf->putSamples(lbuffer, info.framecount) > 0;
				}
				if(sink)
					sink->putBuffer(buffer, n);
				else 
					postAudio(buffer);
				leave();
				Thread::yield();
			}
			else
			{
				memset(&event, 0, sizeof(event));
				event.id = AUDIO_STOP;
				session->queEvent(&event);
				first = true;
			}
		}
		else if(!first)
			--jitter;

                timer = getTimer();

		if(timer > info.framing + 5)
			timer = info.framing + 5;
		if(timer > 5)
                        Thread::sleep(timer - 5);
		else
			Thread::yield();

		enter();
		if(fcount > 1)
			--fcount;
		else if(fcount == 1)
		{
			fcount = 0;
			source = NULL;
			sink = NULL;
			tone = NULL;
			memset(&event, 0, sizeof(event));
			event.id = AUDIO_IDLE;
			session->queEvent(&event);
		}
		leave();		
        }
}

ssize_t RTPStream::putBuffer(Encoded data, size_t len)
{
        uint32 result = info.framing;
        result *= (getCurrentRTPClockRate() / 1000);
        if(!oTimestamp)
                oTimestamp = getCurrentTimestamp();
        oTimestamp += result;
        putData(oTimestamp, (const unsigned char *)data, len);
        return (ssize_t)len;
}

void RTPStream::put2833(struct dtmf2833 *data)
{
	uint32 result = info.framing;
        result *= (getCurrentRTPClockRate() / 1000);
        if(!oTimestamp)
                oTimestamp = getCurrentTimestamp();
        oTimestamp += result;
        putData(oTimestamp, (const unsigned char *)data, sizeof(struct dtmf2833));
}

ssize_t RTPStream::getBuffer(Encoded data, size_t len)
{
        const AppDataUnit *adu;
        size_t n;

        adu = getData(getFirstTimestamp());
        if(!adu)
                return 0;
        n = adu->getSize();
        if(n <= 0)
        {
                delete adu;
                return 0;
        }

        ::memcpy(data, adu->getData(), n);
        delete adu;
        return (ssize_t)n;
}

} // namespace