channel.cpp.svn-base

这是和p2p相关的一份源码

	}
}
// -----------------------------------
void Channel::readMOV()
{
	ChanPacket pack;

	bool done=false;
	headMeta.len = 0;
	while ((!input->eof() && (thread.active)) && (!done))
	{


		unsigned int olen = input->readLong();
		unsigned int otag = input->readLong();

        // don't ask.
		unsigned int len = SWAP4(olen);
		unsigned int tag = SWAP4(otag);

		len -= 8;


		if ((tag == 'mdat') && (len))
		{
			LOG_CHANNEL("Ch.%d mdat: %d",index,len);
			{
				LOG_CHANNEL("mov wait: %d",headMeta.len);

				headMeta.addMem(&olen,4);
				headMeta.addMem(&otag,4);

				while ((!numListeners) && (thread.active))
					sys->sleep(1000);
				sys->sleep(1000);
				LOG_CHANNEL("mov go");

				while ((len) && (thread.active))
				{
					syncPos++;
					pack.init('SYNC',&syncPos,sizeof(syncPos));
					chanData.writePacket(pack);

					int rlen = 4000;
					rlen = input->read(pack.data,rlen);
					pack.type = 'DATA';
					pack.len = rlen;

					chanData.writePacket(pack);
					checkReadDelay(pack.len);

					len -= rlen;

					if (checkBump())
						done=true;
					if (checkIdle())
						done=true;


				}
				LOG_CHANNEL("Mov data end");
			}
		}else{
			LOG_CHANNEL("Ch.%d %c%c%c%c: %d",index,tag>>24&0xff,tag>>16&0xff,tag>>8&0xff,tag&0xff,len);

			if (headMeta.len+len > ChanMeta::MAX_DATALEN)
				throw StreamException("MOV section too big");

			headMeta.addMem(&olen,4);
			headMeta.addMem(&otag,4);
			if (len)
			{
				input->read(headMeta.data+headMeta.len,len);
				headMeta.len += len;
			}
		}
	}


}
// -----------------------------------
bool OggPacket::isBOS()
{
	return (data[5] & 0x02) != 0;
}
// -----------------------------------
bool OggPacket::isNewPacket()
{
	return (data[5] & 0x01) == 0;
}
// -----------------------------------
void OggPacket::read(Stream &in)
{
	// skip until we get OGG capture pattern
	bool gotOgg=false;
	while (!gotOgg)
	{
		if (in.readChar() == 'O')
			if (in.readChar() == 'g')
				if (in.readChar() == 'g')
					if (in.readChar() == 'S')
						gotOgg = true;
		if (!gotOgg)
			LOG_CHANNEL("Skipping OGG packet");
	}



	memcpy(&data[0],"OggS",4);

	in.read(&data[4],27-4);

	int numSegs = data[26];
	bodyLen = 0;

	// read segment table
	in.read(&data[27],numSegs);
	for(int i=0; i<numSegs; i++)
		bodyLen += data[27+i];

	if (bodyLen >= MAX_BODYLEN)		
		throw StreamException("OGG body too big");

	headLen = 27+numSegs;

	if (headLen > MAX_HEADERLEN)
		throw StreamException("OGG header too big");

	in.read(&data[headLen],bodyLen);



	#if 0
		LOG_DEBUG("OGG Packet - page %d, id = %x - %s %s %s - %d segs, %d bytes",
			*(unsigned int *)&data[18],
			*(unsigned int *)&data[14],
			data[5]&0x1?"cont":"new",
			data[5]&0x2?"bos":"",
			data[5]&0x4?"eos":"",
			numSegs,
			headLen+bodyLen);
	#endif

}
// -----------------------------------
void ChanPacket::init(unsigned int t, const void *p, unsigned int l)
{
	type = t;
	if (l > MAX_DATALEN)
		l = MAX_DATALEN;
	len = l;
	memcpy(data,p,len);
}
// -----------------------------------
void ChanPacket::write(Stream &out)
{
	out.writeTag(type);
	out.writeShort(len);
	out.writeShort(0);
	out.write(data,len);
}
// -----------------------------------
void ChanPacket::read(Stream &in)
{
	type = in.readTag();
	len = in.readShort();
	in.readShort();
	if (len > MAX_DATALEN)
		throw StreamException("Bad ChanPacket");
	in.read(data,len);
}

#if 0
// -----------------------------------
unsigned int ChanBuffer::read(unsigned int gpos,void *p,unsigned int len)
{
	if (gpos == 0)
		gpos = pos;

	while (len)
	{
		// skip forward if too far behind
		//if ((pos > MAX_DATALEN) && (gpos < (pos-MAX_DATALEN)))
		//	gpos = pos-MAX_DATALEN;

		// sleep if no data waiting
		unsigned int tim = sys->getTime();
		while (gpos >= pos)
		{
			sys->sleepIdle();
			if ((sys->getTime() - tim) > 30)
				throw TimeoutException();
		}



		int spos = gpos % MAX_DATALEN; // start
		int epos = pos % MAX_DATALEN;	// end

		unsigned int rlen;

		if (pos >= MAX_DATALEN)
		{
			if (spos < epos)	
				rlen = epos-spos;
			else
				rlen = MAX_DATALEN-spos;
		}else{
			rlen = epos-spos;
		}

		// read as many bytes as we can
		if (rlen > len)
			rlen = len;

		if (rlen)
		{

			//LOG("BUF READ gpos=%d, pos=%d, spos=%d, epos=%d, avail=%d",gpos,pos,spos,epos,avail);

			if (rlen > MAX_DATALEN)	
			{
				// something went horribly wrong
				throw StreamException("Chan buffer read fail");
			}

			memcpy(p,&data[spos],rlen);

			p = (char *)p+rlen;
			len -= rlen;
			gpos += rlen;
		}


	}
	return gpos;
}
// -----------------------------------
void	ChanBuffer::write(const void *p,int len)
{
	while (len)
	{
		unsigned int bpos = pos % MAX_DATALEN;
		int wlen = len;
		if ((bpos+wlen) > MAX_DATALEN)
			wlen = MAX_DATALEN-bpos;
		memcpy(&data[bpos],p,wlen);

		p = (char *)p+wlen;
		len -= wlen;
		pos += wlen;
	}
}
// -----------------------------------
void	ChanBuffer::writePacket(ChanPacket &pack)
{
	lock.on();

	unsigned int t = sys->gtohl( SWAP4(pack.type) );
	
	//LOG("write packet %x,%d",t,pack.len);

	lastPacket = pos;
	
	if ((lastPacket-firstPacket) > MAX_DATALEN)
		firstPacket = lastPacket;

	write(&t,4);
	write(&pack.len,sizeof(int));
	write(pack.data,pack.len);

	lock.off();
}
#endif

// -----------------------------------
unsigned int ChanPacketBuffer::writePacket(ChanPacket &pack)
{
	lock.on();

	packets[currPacket%MAX_PACKETS] = pack;
	lastPacket = currPacket;
	currPacket++;

	if (currPacket <= MAX_PACKETS)
		firstPacket = 0;
	else
		firstPacket = currPacket-MAX_PACKETS;

	lock.off();

	return currPacket-1;
}
// -----------------------------------
unsigned int	ChanPacketBuffer::readPacket(unsigned int pos,ChanPacket &pack)
{
	unsigned int tim = sys->getTime();

	if (pos > currPacket)
		pos = currPacket;

	while (pos == currPacket)
	{
		sys->sleepIdle();
		if ((sys->getTime() - tim) > 30)
			throw TimeoutException();
	}

	lock.on();

	if (pos == 0)			// start of stream
		pos = firstPacket;	// or lastPacket
	else if (pos < firstPacket)	// too far behind
		pos = firstPacket;		// so skip forward


	pack = 	packets[pos%MAX_PACKETS];
	pos++;

	lock.off();
	return pos;
}

#if 0
// -----------------------------------
unsigned int	ChanBuffer::readPacket(unsigned int gpos, ChanPacket &pack)
{
	while (gpos >= pos)
		sys->sleepIdle();

	lock.on();

	try
	{

		gpos = read(gpos,&pack.type,sizeof(pack.type));
		pack.type = sys->gtohl( SWAP4(pack.type) );

		gpos = read(gpos,&pack.len,sizeof(pack.len));

		if (pack.len > ChanPacket::MAX_DATALEN)
			pack.len = ChanPacket::MAX_DATALEN;

		gpos = read(gpos,pack.data,pack.len);
	}catch (StreamException &)
	{
	}
	lock.off();
	return gpos;
}

#endif

// -----------------------------------
void Channel::readVorbisIdent(Stream &in)
{
	int ver = in.readLong();
	int chans = in.readChar();
	int rate = in.readLong();
	int brMax = in.readLong();
	int brNom = in.readLong();
	int brLow = in.readLong();

	in.readChar();	// skip blocksize 0+1

	LOG_CHANNEL("OGG Ident: ver=%d, chans=%d, rate=%d, brMax=%d, brNom=%d, brLow=%d",
		ver,chans,rate,brMax,brNom,brLow);

	if (info.bitrate == 0)
	{
		if (brNom > 0)
			info.bitrate = brNom/1000;
		//else
		//	info.bitrate = 0;
	}

	char frame = in.readChar();		// framing bit
	if (!frame)
		throw StreamException("Bad Indent frame");

}

 
// -----------------------------------
void Channel::readVorbisSetup(Stream &in)
{
	// skip everything in packet
	int cnt=0;
	while (!in.eof())
	{
		cnt++;
		in.readChar();
	}

	LOG_CHANNEL("Read %d bytes of Vorbis Setup",cnt);
}
// -----------------------------------
void Channel::readVorbisComment(Stream &in)
{
	int vLen = in.readLong();	// vendor len

	in.skip(vLen);

	char argBuf[8192];

	info.track.clear();

	int cLen = in.readLong();	// comment len
	for(int i=0; i<cLen; i++)
	{
		int l = in.readLong();
		if (l > sizeof(argBuf))
			throw StreamException("Comment string too long");
		in.read(argBuf,l);
		argBuf[l] = 0;
		LOG_CHANNEL("OGG Comment: %s",argBuf);

		char *arg;
		if ((arg=stristr(argBuf,"ARTIST=")))
			info.track.artist.set(arg+7,String::T_ASCII);
		else if ((arg=stristr(argBuf,"TITLE=")))
			info.track.title.set(arg+6,String::T_ASCII);
		else if ((arg=stristr(argBuf,"GENRE=")))
			info.track.genre.set(arg+6,String::T_ASCII);
		else if ((arg=stristr(argBuf,"CONTACT=")))
			info.track.contact.set(arg+8,String::T_ASCII);
		else if ((arg=stristr(argBuf,"ALBUM=")))
			info.track.album.set(arg+6,String::T_ASCII);
	}

	char frame = in.readChar();		// framing bit
	if (!frame)
		throw StreamException("Bad Comment frame");

	updateMeta();

}

// -----------------------------------
void Channel::getStreamPath(char *str)
{
	char idStr[64];

	getIDStr(idStr);

	sprintf(str,"/stream/%s.%s",idStr,info.getTypeExt(info.contentType));
}
// -----------------------------------
void Channel::readVorbisHeaders(Stream &in, OggPacket &ogg)
{
	LOG_CHANNEL("Reading OGG Vorbis headers");
		
//	if (!ogg.isVorbisPacket())
//		throw StreamException("Not Vorbis stream");

	headMeta.len = 0;

	VorbisPacket vorbis;
	

	int numHeaders = 0;
	
	while (numHeaders < 3)	// ogg vorbis always starts with 3 headers
	{

		// read until we have complete ogg page
		vorbis.bodyLen = 0;
		while (!in.eof())
		{
			if (!thread.active)
				throw StreamException("Thread ended waiting for OGG packet");

			if ((vorbis.bodyLen + ogg.bodyLen) >= VorbisPacket::MAX_BODYLEN)
				throw StreamException("Vorbis body too big");

			if (headMeta.len+(ogg.bodyLen+ogg.headLen) >= ChanMeta::MAX_DATALEN)
				throw StreamException("OGG packet too big for headMeta");

			// copy complete ogg packet into headMeta
			memcpy(&headMeta.data[headMeta.len],ogg.data,ogg.headLen+ogg.bodyLen);
			headMeta.len += ogg.headLen+ogg.bodyLen;

			// add body to vorbis packet
			memcpy(&vorbis.body[vorbis.bodyLen],&ogg.data[ogg.headLen],ogg.bodyLen);
			vorbis.bodyLen += ogg.bodyLen;
		
			// read new ogg packet and check for end of last packet.
			ogg.read(in);
			if (ogg.isNewPacket())
				break;
		}

		MemoryStream vin(vorbis.body,vorbis.bodyLen);

		while (!vin.eof())
		{
			char id[7];

			vin.read(id,7);
			if (memcmp(&id[1],"vorbis",6)!=0)
				throw StreamException("Expected Vorbis packet");

			switch (id[0])
			{
				case 1:	// ident
					LOG_CHANNEL("Vorbis Header: Ident");
					readVorbisIdent(vin);
					break;
				case 3: // comment
					LOG_CHANNEL("Vorbis Header: Comment");
					readVorbisComment(vin);
					break;
				case 5: // setup
					LOG_CHANNEL("Vorbis Header: Setup");
					readVorbisSetup(vin);
					break;
				default:
					throw StreamException("Unknown Vorbis packet while reading headers");
					break;
			}

			numHeaders++;
		}

	}


	ChanPacket pack;
	pack.init('HEAD',headMeta.data,headMeta.len);
	headMeta.startPos = chanData.writePacket(pack);

	LOG_CHANNEL("Got Vorbis headers at %d: total %d bytes",headMeta.startPos,headMeta.len);

}
// -----------------------------------
void Channel::readOGG()
{

	OggPacket ogg;
	ChanPacket pack;

	while ((!input->eof() && (thread.active)))
	{

		ogg.read(*input);

		if (ogg.isBOS())
			readVorbisHeaders(*input,ogg);

		syncPos++;
		pack.init('SYNC',&syncPos,sizeof(syncPos));
		chanData.writePacket(pack);

		pack.init('DATA',ogg.data,ogg.headLen+ogg.bodyLen);
		chanData.writePacket(pack);
		checkReadDelay(pack.len);

		if (checkBump())
			throw StreamException("Bumped");
		if (checkIdle())
			break;


	}
}


// -----------------------------------
ASFStream parseASFHeader(Stream &in)
{
	ASFStream asf;

	try
	{
		int numHeaders = in.readLong();

		in.readChar();
		in.readChar();

		LOG_CHANNEL("ASF Headers: %d",numHeaders);
		for(int i=0; i<numHeaders; i++)
		{

			ASFObject obj;

			unsigned int l = obj.readHead(in);
			obj.readData(in,l);


			MemoryStream data(obj.data,obj.lenLo);

			if (obj.type == ASFObject::T_FILE_PROP)
			{
				data.skip(32);

				unsigned int dpLo = data.readLong();
				unsigned int dpHi = data.readLong();

				data.skip(24);

				data.readLong();
				//data.writeLong(1);	// flags = broadcast, not seekable

				int min = data.readLong();
				int max = data.readLong();
				int br = data.readLong();

				if (min != max)
					throw StreamException("ASF packetsizes (min/max) must match");

				asf.packetSize = max;
				asf.bitrate = br;
				asf.numPackets = dpLo;

			}

		}
	}catch(StreamException &e)
	{
		LOG_ERROR("ASF: %s",e.msg);
	}

	return asf;
}
// -----------------------------------
class ASFChunk
{
public:

	void read(Stream &in)
	{
		type = in.readShort();
		len = in.readShort();
		seq = in.readLong();
		v1 = in.readShort();
		v2 = in.readShort();

		dataLen = len-8;
		if (dataLen > sizeof(data)) 
			throw StreamException("ASF chunk too big");
		in.read(data,dataLen);
	}

	void write(Stream &out)
	{
		out.writeShort(type);
		out.writeShort(len);
		out.writeLong(seq);
		out.writeShort(v1);
		out.writeShort(v2);
		out.write(data,dataLen);
	}

	unsigned int seq,dataLen;
	unsigned short type,len,v1,v2;
	unsigned char data[8192];
};