tomwalters@570: /*
tomwalters@570: 	oscpack -- Open Sound Control packet manipulation library
tomwalters@570: 	http://www.audiomulch.com/~rossb/oscpack
tomwalters@570: 
tomwalters@570: 	Copyright (c) 2004-2005 Ross Bencina <rossb@audiomulch.com>
tomwalters@570: 
tomwalters@570: 	Permission is hereby granted, free of charge, to any person obtaining
tomwalters@570: 	a copy of this software and associated documentation files
tomwalters@570: 	(the "Software"), to deal in the Software without restriction,
tomwalters@570: 	including without limitation the rights to use, copy, modify, merge,
tomwalters@570: 	publish, distribute, sublicense, and/or sell copies of the Software,
tomwalters@570: 	and to permit persons to whom the Software is furnished to do so,
tomwalters@570: 	subject to the following conditions:
tomwalters@570: 
tomwalters@570: 	The above copyright notice and this permission notice shall be
tomwalters@570: 	included in all copies or substantial portions of the Software.
tomwalters@570: 
tomwalters@570: 	Any person wishing to distribute modifications to the Software is
tomwalters@570: 	requested to send the modifications to the original developer so that
tomwalters@570: 	they can be incorporated into the canonical version.
tomwalters@570: 
tomwalters@570: 	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
tomwalters@570: 	EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
tomwalters@570: 	MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
tomwalters@570: 	IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR
tomwalters@570: 	ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
tomwalters@570: 	CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
tomwalters@570: 	WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
tomwalters@570: */
tomwalters@570: #include "ip/UdpSocket.h"
tomwalters@570: 
tomwalters@570: #include <vector>
tomwalters@570: #include <algorithm>
tomwalters@570: #include <stdexcept>
tomwalters@570: #include <assert.h>
tomwalters@570: #include <signal.h>
tomwalters@570: #include <math.h>
tomwalters@570: #include <errno.h>
tomwalters@570: #include <string.h> // for memset
tomwalters@570: 
tomwalters@570: #include <pthread.h>
tomwalters@570: #include <unistd.h>
tomwalters@570: #include <stdlib.h>
tomwalters@570: #include <stdio.h>
tomwalters@570: #include <netdb.h>
tomwalters@570: #include <sys/types.h>
tomwalters@570: #include <sys/socket.h>
tomwalters@570: #include <sys/time.h>
tomwalters@570: #include <netinet/in.h> // for sockaddr_in
tomwalters@570: 
tomwalters@570: #include "ip/PacketListener.h"
tomwalters@570: #include "ip/TimerListener.h"
tomwalters@570: 
tomwalters@570: 
tomwalters@570: #if defined(__APPLE__) && !defined(_SOCKLEN_T)
tomwalters@570: // pre system 10.3 didn have socklen_t
tomwalters@570: typedef ssize_t socklen_t;
tomwalters@570: #endif
tomwalters@570: 
tomwalters@570: 
tomwalters@570: static void SockaddrFromIpEndpointName( struct sockaddr_in& sockAddr, const IpEndpointName& endpoint )
tomwalters@570: {
tomwalters@570:     memset( (char *)&sockAddr, 0, sizeof(sockAddr ) );
tomwalters@570:     sockAddr.sin_family = AF_INET;
tomwalters@570: 
tomwalters@570: 	sockAddr.sin_addr.s_addr = 
tomwalters@570: 		(endpoint.address == IpEndpointName::ANY_ADDRESS)
tomwalters@570: 		? INADDR_ANY
tomwalters@570: 		: htonl( endpoint.address );
tomwalters@570: 
tomwalters@570: 	sockAddr.sin_port =
tomwalters@570: 		(endpoint.port == IpEndpointName::ANY_PORT)
tomwalters@570: 		? 0
tomwalters@570: 		: htons( endpoint.port );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: 
tomwalters@570: static IpEndpointName IpEndpointNameFromSockaddr( const struct sockaddr_in& sockAddr )
tomwalters@570: {
tomwalters@570: 	return IpEndpointName( 
tomwalters@570: 		(sockAddr.sin_addr.s_addr == INADDR_ANY) 
tomwalters@570: 			? IpEndpointName::ANY_ADDRESS 
tomwalters@570: 			: ntohl( sockAddr.sin_addr.s_addr ),
tomwalters@570: 		(sockAddr.sin_port == 0)
tomwalters@570: 			? IpEndpointName::ANY_PORT
tomwalters@570: 			: ntohs( sockAddr.sin_port )
tomwalters@570: 		);
tomwalters@570: }
tomwalters@570: 
tomwalters@570: 
tomwalters@570: class UdpSocket::Implementation{
tomwalters@570: 	bool isBound_;
tomwalters@570: 	bool isConnected_;
tomwalters@570: 
tomwalters@570: 	int socket_;
tomwalters@570: 	struct sockaddr_in connectedAddr_;
tomwalters@570: 	struct sockaddr_in sendToAddr_;
tomwalters@570: 
tomwalters@570: public:
tomwalters@570: 
tomwalters@570: 	Implementation()
tomwalters@570: 		: isBound_( false )
tomwalters@570: 		, isConnected_( false )
tomwalters@570: 		, socket_( -1 )
tomwalters@570: 	{
tomwalters@570: 		if( (socket_ = socket( AF_INET, SOCK_DGRAM, 0 )) == -1 ){
tomwalters@570:             throw std::runtime_error("unable to create udp socket\n");
tomwalters@570:         }
tomwalters@570: 
tomwalters@570: 		memset( &sendToAddr_, 0, sizeof(sendToAddr_) );
tomwalters@570:         sendToAddr_.sin_family = AF_INET;
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	~Implementation()
tomwalters@570: 	{
tomwalters@570: 		if (socket_ != -1) close(socket_);
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	IpEndpointName LocalEndpointFor( const IpEndpointName& remoteEndpoint ) const
tomwalters@570: 	{
tomwalters@570: 		assert( isBound_ );
tomwalters@570: 
tomwalters@570: 		// first connect the socket to the remote server
tomwalters@570:         
tomwalters@570:         struct sockaddr_in connectSockAddr;
tomwalters@570: 		SockaddrFromIpEndpointName( connectSockAddr, remoteEndpoint );
tomwalters@570:        
tomwalters@570:         if (connect(socket_, (struct sockaddr *)&connectSockAddr, sizeof(connectSockAddr)) < 0) {
tomwalters@570:             throw std::runtime_error("unable to connect udp socket\n");
tomwalters@570:         }
tomwalters@570: 
tomwalters@570:         // get the address
tomwalters@570: 
tomwalters@570:         struct sockaddr_in sockAddr;
tomwalters@570:         memset( (char *)&sockAddr, 0, sizeof(sockAddr ) );
tomwalters@570:         socklen_t length = sizeof(sockAddr);
tomwalters@570:         if (getsockname(socket_, (struct sockaddr *)&sockAddr, &length) < 0) {
tomwalters@570:             throw std::runtime_error("unable to getsockname\n");
tomwalters@570:         }
tomwalters@570:         
tomwalters@570: 		if( isConnected_ ){
tomwalters@570: 			// reconnect to the connected address
tomwalters@570: 			
tomwalters@570: 			if (connect(socket_, (struct sockaddr *)&connectedAddr_, sizeof(connectedAddr_)) < 0) {
tomwalters@570: 				throw std::runtime_error("unable to connect udp socket\n");
tomwalters@570: 			}
tomwalters@570: 
tomwalters@570: 		}else{
tomwalters@570: 			// unconnect from the remote address
tomwalters@570: 		
tomwalters@570: 			struct sockaddr_in unconnectSockAddr;
tomwalters@570: 			memset( (char *)&unconnectSockAddr, 0, sizeof(unconnectSockAddr ) );
tomwalters@570: 			unconnectSockAddr.sin_family = AF_UNSPEC;
tomwalters@570: 			// address fields are zero
tomwalters@570: 			int connectResult = connect(socket_, (struct sockaddr *)&unconnectSockAddr, sizeof(unconnectSockAddr));
tomwalters@570: 			if ( connectResult < 0 && errno != EAFNOSUPPORT ) {
tomwalters@570: 				throw std::runtime_error("unable to un-connect udp socket\n");
tomwalters@570: 			}
tomwalters@570: 		}
tomwalters@570: 
tomwalters@570: 		return IpEndpointNameFromSockaddr( sockAddr );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	void Connect( const IpEndpointName& remoteEndpoint )
tomwalters@570: 	{
tomwalters@570: 		SockaddrFromIpEndpointName( connectedAddr_, remoteEndpoint );
tomwalters@570:        
tomwalters@570:         if (connect(socket_, (struct sockaddr *)&connectedAddr_, sizeof(connectedAddr_)) < 0) {
tomwalters@570:             throw std::runtime_error("unable to connect udp socket\n");
tomwalters@570:         }
tomwalters@570: 
tomwalters@570: 		isConnected_ = true;
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	void Send( const char *data, int size )
tomwalters@570: 	{
tomwalters@570: 		assert( isConnected_ );
tomwalters@570: 
tomwalters@570:         send( socket_, data, size, 0 );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void SendTo( const IpEndpointName& remoteEndpoint, const char *data, int size )
tomwalters@570: 	{
tomwalters@570: 		sendToAddr_.sin_addr.s_addr = htonl( remoteEndpoint.address );
tomwalters@570:         sendToAddr_.sin_port = htons( remoteEndpoint.port );
tomwalters@570: 
tomwalters@570:         sendto( socket_, data, size, 0, (sockaddr*)&sendToAddr_, sizeof(sendToAddr_) );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	void Bind( const IpEndpointName& localEndpoint )
tomwalters@570: 	{
tomwalters@570: 		struct sockaddr_in bindSockAddr;
tomwalters@570: 		SockaddrFromIpEndpointName( bindSockAddr, localEndpoint );
tomwalters@570: 
tomwalters@570:         if (bind(socket_, (struct sockaddr *)&bindSockAddr, sizeof(bindSockAddr)) < 0) {
tomwalters@570:             throw std::runtime_error("unable to bind udp socket\n");
tomwalters@570:         }
tomwalters@570: 
tomwalters@570: 		isBound_ = true;
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	bool IsBound() const { return isBound_; }
tomwalters@570: 
tomwalters@570:     int ReceiveFrom( IpEndpointName& remoteEndpoint, char *data, int size )
tomwalters@570: 	{
tomwalters@570: 		assert( isBound_ );
tomwalters@570: 
tomwalters@570: 		struct sockaddr_in fromAddr;
tomwalters@570:         socklen_t fromAddrLen = sizeof(fromAddr);
tomwalters@570:              	 
tomwalters@570:         int result = recvfrom(socket_, data, size, 0,
tomwalters@570:                     (struct sockaddr *) &fromAddr, (socklen_t*)&fromAddrLen);
tomwalters@570: 		if( result < 0 )
tomwalters@570: 			return 0;
tomwalters@570: 
tomwalters@570: 		remoteEndpoint.address = ntohl(fromAddr.sin_addr.s_addr);
tomwalters@570: 		remoteEndpoint.port = ntohs(fromAddr.sin_port);
tomwalters@570: 
tomwalters@570: 		return result;
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	int Socket() { return socket_; }
tomwalters@570: };
tomwalters@570: 
tomwalters@570: UdpSocket::UdpSocket()
tomwalters@570: {
tomwalters@570: 	impl_ = new Implementation();
tomwalters@570: }
tomwalters@570: 
tomwalters@570: UdpSocket::~UdpSocket()
tomwalters@570: {
tomwalters@570: 	delete impl_;
tomwalters@570: }
tomwalters@570: 
tomwalters@570: IpEndpointName UdpSocket::LocalEndpointFor( const IpEndpointName& remoteEndpoint ) const
tomwalters@570: {
tomwalters@570: 	return impl_->LocalEndpointFor( remoteEndpoint );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void UdpSocket::Connect( const IpEndpointName& remoteEndpoint )
tomwalters@570: {
tomwalters@570: 	impl_->Connect( remoteEndpoint );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void UdpSocket::Send( const char *data, int size )
tomwalters@570: {
tomwalters@570: 	impl_->Send( data, size );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void UdpSocket::SendTo( const IpEndpointName& remoteEndpoint, const char *data, int size )
tomwalters@570: {
tomwalters@570: 	impl_->SendTo( remoteEndpoint, data, size );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void UdpSocket::Bind( const IpEndpointName& localEndpoint )
tomwalters@570: {
tomwalters@570: 	impl_->Bind( localEndpoint );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: bool UdpSocket::IsBound() const
tomwalters@570: {
tomwalters@570: 	return impl_->IsBound();
tomwalters@570: }
tomwalters@570: 
tomwalters@570: int UdpSocket::ReceiveFrom( IpEndpointName& remoteEndpoint, char *data, int size )
tomwalters@570: {
tomwalters@570: 	return impl_->ReceiveFrom( remoteEndpoint, data, size );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: 
tomwalters@570: struct AttachedTimerListener{
tomwalters@570: 	AttachedTimerListener( int id, int p, TimerListener *tl )
tomwalters@570: 		: initialDelayMs( id )
tomwalters@570: 		, periodMs( p )
tomwalters@570: 		, listener( tl ) {}
tomwalters@570: 	int initialDelayMs;
tomwalters@570: 	int periodMs;
tomwalters@570: 	TimerListener *listener;
tomwalters@570: };
tomwalters@570: 
tomwalters@570: 
tomwalters@570: static bool CompareScheduledTimerCalls( 
tomwalters@570: 		const std::pair< double, AttachedTimerListener > & lhs, const std::pair< double, AttachedTimerListener > & rhs )
tomwalters@570: {
tomwalters@570: 	return lhs.first < rhs.first;
tomwalters@570: }
tomwalters@570: 
tomwalters@570: 
tomwalters@570: SocketReceiveMultiplexer *multiplexerInstanceToAbortWithSigInt_ = 0;
tomwalters@570: 
tomwalters@570: extern "C" /*static*/ void InterruptSignalHandler( int );
tomwalters@570: /*static*/ void InterruptSignalHandler( int )
tomwalters@570: {
tomwalters@570: 	multiplexerInstanceToAbortWithSigInt_->AsynchronousBreak();
tomwalters@570: 	signal( SIGINT, SIG_DFL );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: 
tomwalters@570: class SocketReceiveMultiplexer::Implementation{
tomwalters@570: 	std::vector< std::pair< PacketListener*, UdpSocket* > > socketListeners_;
tomwalters@570: 	std::vector< AttachedTimerListener > timerListeners_;
tomwalters@570: 
tomwalters@570: 	volatile bool break_;
tomwalters@570: 	int breakPipe_[2]; // [0] is the reader descriptor and [1] the writer
tomwalters@570: 
tomwalters@570: 	double GetCurrentTimeMs() const
tomwalters@570: 	{
tomwalters@570: 		struct timeval t;
tomwalters@570: 
tomwalters@570: 		gettimeofday( &t, 0 );
tomwalters@570: 
tomwalters@570: 		return ((double)t.tv_sec*1000.) + ((double)t.tv_usec / 1000.);
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: public:
tomwalters@570:     Implementation()
tomwalters@570: 	{
tomwalters@570: 		if( pipe(breakPipe_) != 0 )
tomwalters@570: 			throw std::runtime_error( "creation of asynchronous break pipes failed\n" );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     ~Implementation()
tomwalters@570: 	{
tomwalters@570: 		close( breakPipe_[0] );
tomwalters@570: 		close( breakPipe_[1] );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void AttachSocketListener( UdpSocket *socket, PacketListener *listener )
tomwalters@570: 	{
tomwalters@570: 		assert( std::find( socketListeners_.begin(), socketListeners_.end(), std::make_pair(listener, socket) ) == socketListeners_.end() );
tomwalters@570: 		// we don't check that the same socket has been added multiple times, even though this is an error
tomwalters@570: 		socketListeners_.push_back( std::make_pair( listener, socket ) );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void DetachSocketListener( UdpSocket *socket, PacketListener *listener )
tomwalters@570: 	{
tomwalters@570: 		std::vector< std::pair< PacketListener*, UdpSocket* > >::iterator i = 
tomwalters@570: 				std::find( socketListeners_.begin(), socketListeners_.end(), std::make_pair(listener, socket) );
tomwalters@570: 		assert( i != socketListeners_.end() );
tomwalters@570: 
tomwalters@570: 		socketListeners_.erase( i );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void AttachPeriodicTimerListener( int periodMilliseconds, TimerListener *listener )
tomwalters@570: 	{
tomwalters@570: 		timerListeners_.push_back( AttachedTimerListener( periodMilliseconds, periodMilliseconds, listener ) );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570: 	void AttachPeriodicTimerListener( int initialDelayMilliseconds, int periodMilliseconds, TimerListener *listener )
tomwalters@570: 	{
tomwalters@570: 		timerListeners_.push_back( AttachedTimerListener( initialDelayMilliseconds, periodMilliseconds, listener ) );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void DetachPeriodicTimerListener( TimerListener *listener )
tomwalters@570: 	{
tomwalters@570: 		std::vector< AttachedTimerListener >::iterator i = timerListeners_.begin();
tomwalters@570: 		while( i != timerListeners_.end() ){
tomwalters@570: 			if( i->listener == listener )
tomwalters@570: 				break;
tomwalters@570: 			++i;
tomwalters@570: 		}
tomwalters@570: 
tomwalters@570: 		assert( i != timerListeners_.end() );
tomwalters@570: 
tomwalters@570: 		timerListeners_.erase( i );
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void Run()
tomwalters@570: 	{
tomwalters@570: 		break_ = false;
tomwalters@570: 
tomwalters@570: 		// configure the master fd_set for select()
tomwalters@570: 
tomwalters@570: 		fd_set masterfds, tempfds;
tomwalters@570: 		FD_ZERO( &masterfds );
tomwalters@570: 		FD_ZERO( &tempfds );
tomwalters@570: 		
tomwalters@570: 		// in addition to listening to the inbound sockets we
tomwalters@570: 		// also listen to the asynchronous break pipe, so that AsynchronousBreak()
tomwalters@570: 		// can break us out of select() from another thread.
tomwalters@570: 		FD_SET( breakPipe_[0], &masterfds );
tomwalters@570: 		int fdmax = breakPipe_[0];		
tomwalters@570: 
tomwalters@570: 		for( std::vector< std::pair< PacketListener*, UdpSocket* > >::iterator i = socketListeners_.begin();
tomwalters@570: 				i != socketListeners_.end(); ++i ){
tomwalters@570: 
tomwalters@570: 			if( fdmax < i->second->impl_->Socket() )
tomwalters@570: 				fdmax = i->second->impl_->Socket();
tomwalters@570: 			FD_SET( i->second->impl_->Socket(), &masterfds );
tomwalters@570: 		}
tomwalters@570: 
tomwalters@570: 
tomwalters@570: 		// configure the timer queue
tomwalters@570: 		double currentTimeMs = GetCurrentTimeMs();
tomwalters@570: 
tomwalters@570: 		// expiry time ms, listener
tomwalters@570: 		std::vector< std::pair< double, AttachedTimerListener > > timerQueue_;
tomwalters@570: 		for( std::vector< AttachedTimerListener >::iterator i = timerListeners_.begin();
tomwalters@570: 				i != timerListeners_.end(); ++i )
tomwalters@570: 			timerQueue_.push_back( std::make_pair( currentTimeMs + i->initialDelayMs, *i ) );
tomwalters@570: 		std::sort( timerQueue_.begin(), timerQueue_.end(), CompareScheduledTimerCalls );
tomwalters@570: 
tomwalters@570: 		const int MAX_BUFFER_SIZE = 4098;
tomwalters@570: 		char *data = new char[ MAX_BUFFER_SIZE ];
tomwalters@570: 		IpEndpointName remoteEndpoint;
tomwalters@570: 
tomwalters@570: 		struct timeval timeout;
tomwalters@570: 
tomwalters@570: 		while( !break_ ){
tomwalters@570: 			tempfds = masterfds;
tomwalters@570: 
tomwalters@570: 			struct timeval *timeoutPtr = 0;
tomwalters@570: 			if( !timerQueue_.empty() ){
tomwalters@570: 				double timeoutMs = timerQueue_.front().first - GetCurrentTimeMs();
tomwalters@570: 				if( timeoutMs < 0 )
tomwalters@570: 					timeoutMs = 0;
tomwalters@570: 			
tomwalters@570: 				// 1000000 microseconds in a second
tomwalters@570: 				timeout.tv_sec = (long)(timeoutMs * .001);
tomwalters@570: 				timeout.tv_usec = (long)((timeoutMs - (timeout.tv_sec * 1000)) * 1000);
tomwalters@570: 				timeoutPtr = &timeout;
tomwalters@570: 			}
tomwalters@570: 
tomwalters@570: 			if( select( fdmax + 1, &tempfds, 0, 0, timeoutPtr ) < 0 && errno != EINTR ){
tomwalters@570:    				throw std::runtime_error("select failed\n");
tomwalters@570: 			}
tomwalters@570: 
tomwalters@570: 			if ( FD_ISSET( breakPipe_[0], &tempfds ) ){
tomwalters@570: 				// clear pending data from the asynchronous break pipe
tomwalters@570: 				char c;
tomwalters@570: 				read( breakPipe_[0], &c, 1 );
tomwalters@570: 			}
tomwalters@570: 			
tomwalters@570: 			if( break_ )
tomwalters@570: 				break;
tomwalters@570: 
tomwalters@570: 			for( std::vector< std::pair< PacketListener*, UdpSocket* > >::iterator i = socketListeners_.begin();
tomwalters@570: 					i != socketListeners_.end(); ++i ){
tomwalters@570: 
tomwalters@570: 				if( FD_ISSET( i->second->impl_->Socket(), &tempfds ) ){
tomwalters@570: 
tomwalters@570: 					int size = i->second->ReceiveFrom( remoteEndpoint, data, MAX_BUFFER_SIZE );
tomwalters@570: 					if( size > 0 ){
tomwalters@570: 						i->first->ProcessPacket( data, size, remoteEndpoint );
tomwalters@570: 						if( break_ )
tomwalters@570: 							break;
tomwalters@570: 					}
tomwalters@570: 				}
tomwalters@570: 			}
tomwalters@570: 
tomwalters@570: 			// execute any expired timers
tomwalters@570: 			currentTimeMs = GetCurrentTimeMs();
tomwalters@570: 			bool resort = false;
tomwalters@570: 			for( std::vector< std::pair< double, AttachedTimerListener > >::iterator i = timerQueue_.begin();
tomwalters@570: 					i != timerQueue_.end() && i->first <= currentTimeMs; ++i ){
tomwalters@570: 
tomwalters@570: 				i->second.listener->TimerExpired();
tomwalters@570: 				if( break_ )
tomwalters@570: 					break;
tomwalters@570: 
tomwalters@570: 				i->first += i->second.periodMs;
tomwalters@570: 				resort = true;
tomwalters@570: 			}
tomwalters@570: 			if( resort )
tomwalters@570: 				std::sort( timerQueue_.begin(), timerQueue_.end(), CompareScheduledTimerCalls );
tomwalters@570: 		}
tomwalters@570: 
tomwalters@570: 		delete [] data;
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void Break()
tomwalters@570: 	{
tomwalters@570: 		break_ = true;
tomwalters@570: 	}
tomwalters@570: 
tomwalters@570:     void AsynchronousBreak()
tomwalters@570: 	{
tomwalters@570: 		break_ = true;
tomwalters@570: 
tomwalters@570: 		// Send a termination message to the asynchronous break pipe, so select() will return
tomwalters@570: 		write( breakPipe_[1], "!", 1 );
tomwalters@570: 	}
tomwalters@570: };
tomwalters@570: 
tomwalters@570: 
tomwalters@570: 
tomwalters@570: SocketReceiveMultiplexer::SocketReceiveMultiplexer()
tomwalters@570: {
tomwalters@570: 	impl_ = new Implementation();
tomwalters@570: }
tomwalters@570: 
tomwalters@570: SocketReceiveMultiplexer::~SocketReceiveMultiplexer()
tomwalters@570: {	
tomwalters@570: 	delete impl_;
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::AttachSocketListener( UdpSocket *socket, PacketListener *listener )
tomwalters@570: {
tomwalters@570: 	impl_->AttachSocketListener( socket, listener );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::DetachSocketListener( UdpSocket *socket, PacketListener *listener )
tomwalters@570: {
tomwalters@570: 	impl_->DetachSocketListener( socket, listener );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::AttachPeriodicTimerListener( int periodMilliseconds, TimerListener *listener )
tomwalters@570: {
tomwalters@570: 	impl_->AttachPeriodicTimerListener( periodMilliseconds, listener );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::AttachPeriodicTimerListener( int initialDelayMilliseconds, int periodMilliseconds, TimerListener *listener )
tomwalters@570: {
tomwalters@570: 	impl_->AttachPeriodicTimerListener( initialDelayMilliseconds, periodMilliseconds, listener );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::DetachPeriodicTimerListener( TimerListener *listener )
tomwalters@570: {
tomwalters@570: 	impl_->DetachPeriodicTimerListener( listener );
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::Run()
tomwalters@570: {
tomwalters@570: 	impl_->Run();
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::RunUntilSigInt()
tomwalters@570: {
tomwalters@570: 	assert( multiplexerInstanceToAbortWithSigInt_ == 0 ); /* at present we support only one multiplexer instance running until sig int */
tomwalters@570: 	multiplexerInstanceToAbortWithSigInt_ = this;
tomwalters@570: 	signal( SIGINT, InterruptSignalHandler );
tomwalters@570: 	impl_->Run();
tomwalters@570: 	signal( SIGINT, SIG_DFL );
tomwalters@570: 	multiplexerInstanceToAbortWithSigInt_ = 0;
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::Break()
tomwalters@570: {
tomwalters@570: 	impl_->Break();
tomwalters@570: }
tomwalters@570: 
tomwalters@570: void SocketReceiveMultiplexer::AsynchronousBreak()
tomwalters@570: {
tomwalters@570: 	impl_->AsynchronousBreak();
tomwalters@570: }
tomwalters@570: