packageclient.h

C++ patterns设计模式

#ifndef INCLUDED_PACKAGECLIENT
#define INCLUDED_PACKAGECLIENT

#if defined(HAS_PRAGMA_ONCE)
#pragma PRAGMA_ONCE_DECLARE
#endif

#include "tcpclient.h"
#include "packagebase.h"

namespace acl
{
	template<typename PackageType>
	class PackageClient : public TcpClient
						, public PackageBase<PackageType>
	{
	public:
		PackageClient()
			: PackageBase<PackageType>(QUEUE_SIZE)
		{
		}
		PackageClient(u_short port, const char *addr)
			: PackageBase<PackageType>(QUEUE_SIZE)
		{
		}
		virtual ~PackageClient()
		{
		}
		// convert stream into package, and put into queue
		bool queueStream(protocol::BaseStream &stream)
		{
			PackageType *pkg = new PackageType;
			if(!pkg->fromStream(stream))
				return false;
			putPackageAtTail(pkg);
			return true;
		}

		// send package to peer
		bool sendPackage(PackageType *pkg)
		{			
			assert(pkg);
			protocol::BaseStream stream;
			if(packageWithSize())
			{
				if(!pkg->toStream(stream))
					return false;
			}
			else
			{
				// convert package to stream
				protocol::BaseStream pkgStream;
				int length = 0;
				if(pkg->toStream(pkgStream))
				{
					length = pkgStream.size() - 4;
					// the real stream with length indicator
					stream.enc32(length);
					stream += pkgStream;
				}
				else
				{
					return false;
				}
			}
			return TcpClient::send(stream.getBuffer(), stream.size()) == stream.size();
		}

	private:
        virtual int		onReceive();
		int  recvError(int ret);

	private:
		enum
		{
			QUEUE_SIZE = 4096
		};
		enum 
		{
			RECV_TIMEOUT = 1000
		};
		enum
		{
			MAX_PACKAGE_LENGTH = 4096
		};        

	};


	template<typename PackageType>
		int PackageClient<PackageType>::onReceive()
	{
		protocol::BaseStream stream;
		int  ret;

		//receive the package length at first!
		stream.resize(4);
		ret = m_comm.recvAsyn(stream.getWriteBuffer(), stream.size(), RECV_TIMEOUT);

		//check valid header
		if(ret != 4)
		{
			if(ret > 0)
			{
				trace(m_comm, "WRN:get package length failed, "
					"just received(%d) bytes\n", ret);
			}
			return recvError(ret);
		}
		traceBinary(m_comm, "recv header :\n", stream.getBuffer(), stream.size());

		//convert to host byte order
		u32 length;
		stream.dec32(length);
		stream.reset();			// reset the decode position
		u32 left = length;
		if(packageWithSize())
		{
			left -= 4;
		}
		//package can't just have the length field.
		if(left <= 0 || left >= MAX_PACKAGE_LENGTH)
		{
			trace(m_comm, "WRN:invalid length(%d)\n", length);
			return RECV_USER_ERROR;
		}
		// alloc the new buffer
		stream.resize(length);
		//receive the left package data
		ret = m_comm.recvAsyn(stream.getWriteBuffer(length-left), left, RECV_TIMEOUT);

		if(ret != left)
		{
			trace(m_comm, "get left package failed, want(%d) but received(%d)\n",
				left, ret);

			return recvError(ret);
		}
		//if debug, we print the size and buffer content, else size only
		traceBinary(m_comm, "recv buffer : \n", stream.getBuffer(), stream.size());
		if(!queueStream(stream))
		{
			trace(m_comm, "stream error! I can't create package from it\n");
		}
		return RECV_OK;
	}

	template<typename PackageType>
		int  PackageClient<PackageType>::recvError(int ret)
	{
		if(ret == 0)
			return RECV_CLOSE;
		else if(ret > 0)
			return RECV_USER_ERROR;
		else
			return RECV_SOCKET_ERROR;
	}
}
#endif