msgSender.d

import	std.stdio;
import	std.string;
import	std.socket;
import	std.socketstream;
import	core.thread;
import	core.exception;
import	core.sync.condition;
import	core.sync.mutex;

// Queueテンプレートの定義
mixin	template	Queue(DataType){
	class	Queue{
		DataType[]	dataBuf;
		ulong	size;
		ulong	free;
		ulong	w;
		ulong	r;

		this(ulong	size){
			dataBuf = new DataType[size];
			w = 0;
			r = 0;
			this.size=size;
		}
		void	write(DataType	data){
			if(free>=0){
				dataBuf[w]=data;
				w++;
				free--;
				w%=size;
			}else{
				throw	new	Exception("Queue data full.");
			}
		}
		DataType	read(){
			DataType	data;
			if((size-free)>0){
				data=dataBuf[r];
				r++;
				free++;
				r%=size;
			}else{
				throw	new	Exception("Queue data empty.");
			}
			return	data;
		}
		
		ulong	getDataCount(){
			return	(size-free);
		}
	}
}
// string型のQueueをテンプレートから定義
mixin	Queue!(string) Qstring;

interface	ISocketStreamController{
	void	run();
	void	close();
}

class	SocketRead:Thread,ISocketStreamController{
	SocketStream	ss;
	bool	running=true;
	ubyte[]	buf;
	ulong	readSize;

	this(SocketStream	ss){
		super(&run);
		buf=new ubyte[4096];
		this.ss=ss;
		readSize=0;
	}

	void	run(){
		try{
			while(true){
				readSize=ss.read(buf);
				if(readSize==0){
					break;
				}
				writeln(cast(string)buf);
			}
		}catch(Exception	e){
			writeln(e);
		}catch(Error	e){
			writeln(e);
		}finally{
		}
	}
	void	close(){
		ss.socket.shutdown(SocketShutdown.RECEIVE);
	}
}


class	SocketWrite:Thread,ISocketStreamController{
	SocketStream	ss;
	bool	running=true;
	Mutex	mt;
	Condition	c;
	Qstring.Queue	q;

	this(SocketStream	ss){
		super(&run);
		this.ss=ss;
		mt=new	Mutex;
		c=new	Condition(mt);
		q=new Qstring.Queue(10);
	}

	void	run(){
		try{
			// ループ
			while(running==true){
				if(q.getDataCount()==0){
					synchronized(mt){
						c.wait();
					}
				}else{
					ss.writeString(q.read());
				}
			}
		}catch(Exception	e){
			writeln(e);
		}catch(Error	e){
			writeln(e);
		}finally{
		// 終了処理
		}
	}

	private	void	notify(){
		synchronized(mt){
			c.notify();
		}
	}

	void	sendString(string	s){
		q.write(s);
		notify();
	}

	void	close(){
		running=false;
		notify();
		ss.socket.shutdown(SocketShutdown.SEND);
	}
}

ushort	eNetPort=80;
string	eNetHost="192.168.1.1";

void	main(){
	string	msg;
/*	UdpSocket	udpsock = new UdpSocket();
	udpsock.connect(new InternetAddress(eNetHost,eNetPort));
	SocketStream	sockStream=new SocketStream(udpsock);
*/
	SocketStream	sockStream=new SocketStream(new TcpSocket(new InternetAddress(eNetHost,eNetPort)));
	SocketRead	sr=new	SocketRead(sockStream);
	SocketWrite	sw=new	SocketWrite(sockStream);

	sr.start();
	sw.start();
	writeln("Connected.");
	while(true){
		msg=chomp(readln())~"\r\n";
		if(msg=="quit\r\n"){
			break;
		}
		sw.sendString(msg);
	}
	sr.close();
	sw.close();
	sockStream.socket.close();
	sr.join();
	sw.join();
	return;
}

お勉強がてら書いてみた。