2 // This file is part of the aMule Project.
4 // Copyright (c) 2011-2011 aMule Team ( admin@amule.org / http://www.amule.org )
5 // Copyright (c) 2011-2011 Stu Redman ( admin@amule.org / http://www.amule.org )
7 // Any parts of this program derived from the xMule, lMule or eMule project,
8 // or contributed by third-party developers are copyrighted by their
11 // This program is free software; you can redistribute it and/or modify
12 // it under the terms of the GNU General Public License as published by
13 // the Free Software Foundation; either version 2 of the License, or
14 // (at your option) any later version.
16 // This program is distributed in the hope that it will be useful,
17 // but WITHOUT ANY WARRANTY; without even the implied warranty of
18 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 // GNU General Public License for more details.
21 // You should have received a copy of the GNU General Public License
22 // along with this program; if not, write to the Free Software
23 // Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
26 #include "config.h" // Needed for HAVE_BOOST_SOURCES
30 #define _WIN32_WINNT 0x0501 // Boost complains otherwise
33 // Windows requires that Boost headers are included before wx headers.
34 // This works if precompiled headers are disabled for this file.
36 #define BOOST_ALL_NO_LIB
38 // Suppress warning caused by faulty boost/preprocessor/config/config.hpp in Boost 1.49
39 #if defined __GNUC__ && ! defined __GXX_EXPERIMENTAL_CXX0X__ && __cplusplus < 201103L
40 #define BOOST_PP_VARIADICS 0
43 #include <algorithm> // Needed for std::min - Boost up to 1.54 fails to compile with MSVC 2013 otherwise
45 #include <boost/asio.hpp>
46 #include <boost/bind.hpp>
47 #include <boost/version.hpp>
50 // Do away with building Boost.System, adding lib paths...
51 // Just include the single file and be done.
53 #ifdef HAVE_BOOST_SOURCES
54 # include <boost/../libs/system/src/error_code.cpp>
56 # include <boost/system/error_code.hpp>
59 #include "LibSocket.h"
60 #include <wx/thread.h> // wxMutex
61 #include <wx/intl.h> // _()
62 #include <common/Format.h> // Needed for CFormat
64 #include "GuiEvents.h"
65 #include "amuleIPV4Address.h"
66 #include "MuleUDPSocket.h"
67 #include "OtherFunctions.h" // DeleteContents
68 #include "ScopedPtr.h"
69 #include <common/Macros.h>
71 using namespace boost::asio
;
72 using namespace boost::system
; // for error_code
73 static io_service s_io_service
;
75 // Number of threads in the Asio thread pool
76 const int CAsioService::m_numberOfThreads
= 4;
79 * ASIO Client TCP socket implementation
82 class CamuleIPV4Endpoint
: public ip::tcp::endpoint
{
84 CamuleIPV4Endpoint() {}
86 CamuleIPV4Endpoint(const CamuleIPV4Endpoint
& impl
) : ip::tcp::endpoint(impl
) {}
87 CamuleIPV4Endpoint(const ip::tcp::endpoint
& ep
) { * this = ep
; }
88 CamuleIPV4Endpoint(const ip::udp::endpoint
& ep
) { address(ep
.address()); port(ep
.port()); }
90 const CamuleIPV4Endpoint
& operator = (const ip::tcp::endpoint
& ep
)
92 * (ip::tcp::endpoint
*) this = ep
;
100 // cppcheck-suppress uninitMemberVar m_readBufferPtr
101 CAsioSocketImpl(CLibSocket
* libSocket
) :
102 m_libSocket(libSocket
),
103 m_strand(s_io_service
),
104 m_timer(s_io_service
)
107 m_blocksRead
= false;
108 m_blocksWrite
= false;
111 m_readBufferSize
= 0;
112 m_readPending
= false;
113 m_readBufferContent
= 0;
114 m_eventPending
= false;
119 m_isDestroying
= false;
120 m_proxyState
= false;
125 m_socket
= new ip::tcp::socket(s_io_service
);
127 // Set socket to non blocking
128 m_socket
->non_blocking();
133 delete[] m_readBuffer
;
134 delete[] m_sendBuffer
;
137 void Notify(bool notify
)
142 bool Connect(const amuleIPV4Address
& adr
, bool wait
)
147 m_port
= adr
.Service();
150 m_sync
= !m_notify
; // set this once for the whole lifetime of the socket
151 AddDebugLogLineF(logAsio
, CFormat(wxT("Connect %s %p")) % m_IP
% this);
153 if (wait
|| m_sync
) {
155 m_socket
->connect(adr
.GetEndpoint(), ec
);
160 m_socket
->async_connect(adr
.GetEndpoint(),
161 m_strand
.wrap(boost::bind(& CAsioSocketImpl::HandleConnect
, this, placeholders::error
)));
162 // m_OK and return are false because we are not connected yet
167 bool IsConnected() const
172 // For wxSocketClient, Ok won't return true unless the client is connected to a server.
178 bool IsDestroying() const
180 return m_isDestroying
;
183 // Returns the actual error code
184 int LastError() const
189 // Is reading blocked?
190 bool BlocksRead() const
195 // Is writing blocked?
196 bool BlocksWrite() const
198 return m_blocksWrite
;
201 // Problem: wx sends an event when data gets available, so first there is an event, then Read() is called
202 // Asio can read async with callback, so you first read, then you get an event.
204 // - Read some data in background into a buffer
205 // - Callback posts event when something is there
206 // - Read data from buffer
207 // - If data is exhausted, start reading more in background
208 // - If not, post another event (making sure events don't pile up though)
209 uint32
Read(char * buf
, uint32 bytesToRead
)
211 if (bytesToRead
== 0) { // huh?
216 return ReadSync(buf
, bytesToRead
);
220 AddDebugLogLineF(logAsio
, CFormat(wxT("Read1 %s %d - Error")) % m_IP
% bytesToRead
);
224 if (m_readPending
// Background read hasn't completed.
225 || m_readBufferContent
== 0) { // shouldn't be if it's not pending
228 AddDebugLogLineF(logAsio
, CFormat(wxT("Read1 %s %d - Block")) % m_IP
% bytesToRead
);
232 m_blocksRead
= false; // shouldn't be needed
234 // Read from our buffer
235 uint32 readCache
= std::min(m_readBufferContent
, bytesToRead
);
236 memcpy(buf
, m_readBufferPtr
, readCache
);
237 m_readBufferContent
-= readCache
;
238 m_readBufferPtr
+= readCache
;
240 AddDebugLogLineF(logAsio
, CFormat(wxT("Read2 %s %d - %d")) % m_IP
% bytesToRead
% readCache
);
241 if (m_readBufferContent
) {
242 // Data left, post another event
245 // Nothing left, read more
246 StartBackgroundRead();
252 // Make a copy of the data and send it in background
253 // - unless a background send is already going on
254 uint32
Write(const void * buf
, uint32 nbytes
)
257 return WriteSync(buf
, nbytes
);
261 m_blocksWrite
= true;
262 AddDebugLogLineF(logAsio
, CFormat(wxT("Write blocks %d %p %s")) % nbytes
% m_sendBuffer
% m_IP
);
265 AddDebugLogLineF(logAsio
, CFormat(wxT("Write %d %s")) % nbytes
% m_IP
);
266 m_sendBuffer
= new char[nbytes
];
267 memcpy(m_sendBuffer
, buf
, nbytes
);
268 m_strand
.dispatch(boost::bind(& CAsioSocketImpl::DispatchWrite
, this, nbytes
));
279 if (m_sync
|| s_io_service
.stopped()) {
282 m_strand
.dispatch(boost::bind(& CAsioSocketImpl::DispatchClose
, this));
290 if (m_isDestroying
) {
291 AddDebugLogLineC(logAsio
, CFormat(wxT("Destroy() already dying socket %p %p %s")) % m_libSocket
% this % m_IP
);
294 m_isDestroying
= true;
295 AddDebugLogLineF(logAsio
, CFormat(wxT("Destroy() %p %p %s")) % m_libSocket
% this % m_IP
);
297 if (m_sync
|| s_io_service
.stopped()) {
300 // Close prevents creation of any more callbacks, but does not clear any callbacks already
301 // sitting in Asio's event queue (I have seen such a crash).
302 // So create a delay timer so they can be called until core is notified.
303 m_timer
.expires_from_now(boost::posix_time::seconds(1));
304 m_timer
.async_wait(m_strand
.wrap(boost::bind(& CAsioSocketImpl::HandleDestroy
, this)));
320 // Bind socket to local endpoint if user wants to choose the local address
322 void SetLocal(const amuleIPV4Address
& local
)
325 if (!m_socket
->is_open()) {
326 // Socket is usually still closed when this is called
327 m_socket
->open(boost::asio::ip::tcp::v4(), ec
);
329 AddDebugLogLineC(logAsio
, CFormat(wxT("Can't open socket : %s")) % ec
.message());
333 // We are using random (OS-defined) local ports.
334 // To set a constant output port, first call
335 // m_socket->set_option(socket_base::reuse_address(true));
336 // and then set the endpoint's port to it.
338 CamuleIPV4Endpoint
endpoint(local
.GetEndpoint());
340 m_socket
->bind(endpoint
, ec
);
342 AddDebugLogLineC(logAsio
, CFormat(wxT("Can't bind socket to local endpoint %s : %s"))
343 % local
.IPAddress() % ec
.message());
345 AddDebugLogLineF(logAsio
, CFormat(wxT("Bound socket to local endpoint %s")) % local
.IPAddress());
350 void EventProcessed()
352 m_eventPending
= false;
355 void SetWrapSocket(CLibSocket
* socket
)
357 m_libSocket
= socket
;
358 // Also do some setting up
362 StartBackgroundRead();
368 amuleIPV4Address addr
= CamuleIPV4Endpoint(m_socket
->remote_endpoint(ec
));
370 AddDebugLogLineN(logAsio
, CFormat(wxT("UpdateIP failed %p %s")) % this % ec
.message());
374 m_port
= addr
.Service();
375 AddDebugLogLineF(logAsio
, CFormat(wxT("UpdateIP %s %d %p")) % m_IP
% m_port
% this);
379 const wxChar
* GetIP() const { return m_IP
; }
380 uint16
GetPort() const { return m_port
; }
382 ip::tcp::socket
& GetAsioSocket()
387 bool GetProxyState() const { return m_proxyState
; }
389 void SetProxyState(bool state
, const amuleIPV4Address
* adr
)
391 m_proxyState
= state
;
393 // Start. Get the true IP for logging.
396 AddDebugLogLineF(logAsio
, CFormat(wxT("SetProxyState to proxy %s")) % m_IP
);
398 // Transition from proxy to normal mode
399 AddDebugLogLineF(logAsio
, CFormat(wxT("SetProxyState to normal %s")) % m_IP
);
407 // Access to m_socket is all bundled in the thread running s_io_service to avoid
408 // concurrent access to the socket from several threads.
409 // So once things are running (after connect), all access goes through one of these handlers.
416 AddDebugLogLineC(logAsio
, CFormat(wxT("Close error %s %s")) % m_IP
% ec
.message());
418 AddDebugLogLineF(logAsio
, CFormat(wxT("Closed %s")) % m_IP
);
422 void DispatchBackgroundRead()
424 AddDebugLogLineF(logAsio
, CFormat(wxT("DispatchBackgroundRead %s")) % m_IP
);
425 m_socket
->async_read_some(null_buffers(),
426 m_strand
.wrap(boost::bind(& CAsioSocketImpl::HandleRead
, this, placeholders::error
)));
429 void DispatchWrite(uint32 nbytes
)
431 async_write(*m_socket
, buffer(m_sendBuffer
, nbytes
),
432 m_strand
.wrap(boost::bind(& CAsioSocketImpl::HandleSend
, this, placeholders::error
, placeholders::bytes_transferred
)));
436 // Completion handlers for async requests
439 void HandleConnect(const error_code
& err
)
442 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleConnect %d %s")) % m_OK
% m_IP
);
443 if (m_isDestroying
) {
444 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleConnect: socket pending for deletion %s")) % m_IP
);
446 CoreNotify_LibSocketConnect(m_libSocket
, err
.value());
448 // After connect also send a OUTPUT event to show data is available
449 CoreNotify_LibSocketSend(m_libSocket
, 0);
451 StartBackgroundRead();
457 void HandleSend(const error_code
& err
, size_t DEBUG_ONLY(bytes_transferred
) )
459 delete[] m_sendBuffer
;
462 if (m_isDestroying
) {
463 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleSend: socket pending for deletion %s")) % m_IP
);
466 AddDebugLogLineN(logAsio
, CFormat(wxT("HandleSend Error %d %s")) % bytes_transferred
% m_IP
);
469 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleSend %d %s")) % bytes_transferred
% m_IP
);
470 m_blocksWrite
= false;
471 CoreNotify_LibSocketSend(m_libSocket
, m_ErrorCode
);
476 void HandleRead(const error_code
& ec
)
478 if (m_isDestroying
) {
479 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleRead: socket pending for deletion %s")) % m_IP
);
483 // This is what we get in Windows when a connection gets closed from remote.
484 AddDebugLogLineN(logAsio
, CFormat(wxT("HandleReadError %s %s")) % m_IP
% ec
.message());
490 uint32 avail
= m_socket
->available(ec2
);
492 AddDebugLogLineN(logAsio
, CFormat(wxT("HandleReadError available %d %s %s")) % avail
% m_IP
% ec2
.message());
497 // This is what we get in Linux when a connection gets closed from remote.
498 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleReadError nothing available %s")) % m_IP
);
503 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleRead %d %s")) % avail
% m_IP
);
505 // adjust (or create) our read buffer
506 if (m_readBufferSize
< avail
) {
507 delete[] m_readBuffer
;
508 m_readBuffer
= new char[avail
];
509 m_readBufferSize
= avail
;
511 m_readBufferPtr
= m_readBuffer
;
513 // read available data
514 m_readBufferContent
= m_socket
->read_some(buffer(m_readBuffer
, avail
), ec2
);
515 if (SetError(ec2
) || m_readBufferContent
== 0) {
516 AddDebugLogLineN(logAsio
, CFormat(wxT("HandleReadError read %d %s %s")) % m_readBufferContent
% m_IP
% ec2
.message());
521 m_readPending
= false;
522 m_blocksRead
= false;
528 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleDestroy() %p %p %s")) % m_libSocket
% this % m_IP
);
529 CoreNotify_LibSocketDestroy(m_libSocket
);
537 void StartBackgroundRead()
539 m_readPending
= true;
540 m_readBufferContent
= 0;
541 m_strand
.dispatch(boost::bind(& CAsioSocketImpl::DispatchBackgroundRead
, this));
544 void PostReadEvent(int DEBUG_ONLY(from
) )
546 if (!m_eventPending
) {
547 AddDebugLogLineF(logAsio
, CFormat(wxT("Post read event %d %s")) % from
% m_IP
);
548 m_eventPending
= true;
549 CoreNotify_LibSocketReceive(m_libSocket
, m_ErrorCode
);
555 if (!m_isDestroying
&& !m_closed
) {
556 CoreNotify_LibSocketLost(m_libSocket
);
565 bool SetError(const error_code
& err
)
567 m_ErrorCode
= err
.value();
568 return m_ErrorCode
!= errc::success
;
572 // Synchronous sockets (amulecmd)
574 uint32
ReadSync(char * buf
, uint32 bytesToRead
)
577 uint32 received
= read(*m_socket
, buffer(buf
, bytesToRead
), ec
);
582 uint32
WriteSync(const void * buf
, uint32 nbytes
)
585 uint32 sent
= write(*m_socket
, buffer(buf
, nbytes
), ec
);
591 // Access to even const & wxString is apparently not thread-safe.
592 // Locks are set/removed in wx and reference counts can go astray.
593 // So store our IP string in a wxString which is used nowhere.
594 // Store a pointer to its string buffer as well and use THAT everywhere.
596 void SetIp(const amuleIPV4Address
& adr
)
598 m_IPstring
= adr
.IPAddress();
599 m_IP
= m_IPstring
.c_str();
600 m_IPint
= StringIPtoUint32(m_IPstring
);
603 CLibSocket
* m_libSocket
;
604 ip::tcp::socket
* m_socket
;
606 wxString m_IPstring
; // as String (use nowhere because of threading!)
607 const wxChar
* m_IP
; // as char* (use in debug logs)
608 uint32 m_IPint
; // as int
609 uint16 m_port
; // remote port
615 uint32 m_readBufferSize
;
616 char * m_readBufferPtr
;
618 uint32 m_readBufferContent
;
621 io_service::strand m_strand
; // handle synchronisation in io_service thread pool
622 deadline_timer m_timer
;
625 bool m_isDestroying
; // true if Destroy() was called
627 bool m_notify
; // set by Notify()
628 bool m_sync
; // copied from !m_notify on Connect()
633 * Library socket wrapper
636 CLibSocket::CLibSocket(int /* flags */)
638 m_aSocket
= new CAsioSocketImpl(this);
642 CLibSocket::~CLibSocket()
644 AddDebugLogLineF(logAsio
, CFormat(wxT("~CLibSocket() %p %p %s")) % this % m_aSocket
% m_aSocket
->GetIP());
649 bool CLibSocket::Connect(const amuleIPV4Address
& adr
, bool wait
)
651 return m_aSocket
->Connect(adr
, wait
);
655 bool CLibSocket::IsConnected() const
657 return m_aSocket
->IsConnected();
661 bool CLibSocket::IsOk() const
663 return m_aSocket
->IsOk();
667 wxString
CLibSocket::GetPeer()
669 return m_aSocket
->GetPeer();
673 uint32
CLibSocket::GetPeerInt()
675 return m_aSocket
->GetPeerInt();
679 void CLibSocket::Destroy()
681 m_aSocket
->Destroy();
685 bool CLibSocket::IsDestroying() const
687 return m_aSocket
->IsDestroying();
691 void CLibSocket::Notify(bool notify
)
693 m_aSocket
->Notify(notify
);
697 uint32
CLibSocket::Read(void * buffer
, uint32 nbytes
)
699 return m_aSocket
->Read((char *) buffer
, nbytes
);
703 uint32
CLibSocket::Write(const void * buffer
, uint32 nbytes
)
705 return m_aSocket
->Write(buffer
, nbytes
);
709 void CLibSocket::Close()
715 int CLibSocket::LastError() const
717 return m_aSocket
->LastError();
721 void CLibSocket::SetLocal(const amuleIPV4Address
& local
)
723 m_aSocket
->SetLocal(local
);
730 bool CLibSocket::BlocksRead() const
732 return m_aSocket
->BlocksRead();
736 bool CLibSocket::BlocksWrite() const
738 return m_aSocket
->BlocksWrite();
742 void CLibSocket::EventProcessed()
744 m_aSocket
->EventProcessed();
748 void CLibSocket::LinkSocketImpl(class CAsioSocketImpl
* socket
)
752 m_aSocket
->SetWrapSocket(this);
756 const wxChar
* CLibSocket::GetIP() const
758 return m_aSocket
->GetIP();
762 bool CLibSocket::GetProxyState() const
764 return m_aSocket
->GetProxyState();
768 void CLibSocket::SetProxyState(bool state
, const amuleIPV4Address
* adr
)
770 m_aSocket
->SetProxyState(state
, adr
);
775 * ASIO TCP socket server
778 class CAsioSocketServerImpl
: public ip::tcp::acceptor
781 CAsioSocketServerImpl(const amuleIPV4Address
& adr
, CLibSocketServer
* libSocketServer
)
782 : ip::tcp::acceptor(s_io_service
),
783 m_libSocketServer(libSocketServer
),
784 m_currentSocket(NULL
),
785 m_strand(s_io_service
)
788 m_socketAvailable
= false;
791 open(adr
.GetEndpoint().protocol());
792 set_option(ip::tcp::acceptor::reuse_address(true));
793 bind(adr
.GetEndpoint());
797 AddDebugLogLineN(logAsio
, CFormat(wxT("CAsioSocketServerImpl bind to %s %d")) % adr
.IPAddress() % adr
.Service());
798 } catch (const system_error
& err
) {
799 AddDebugLogLineC(logAsio
, CFormat(wxT("CAsioSocketServerImpl bind to %s %d failed - %s")) % adr
.IPAddress() % adr
.Service() % err
.code().message());
803 ~CAsioSocketServerImpl()
807 // For wxSocketServer, Ok will return true if the server could bind to the specified address and is already listening for new connections.
808 bool IsOk() const { return m_ok
; }
810 void Close() { close(); }
812 bool AcceptWith(CLibSocket
& socket
)
814 if (!m_socketAvailable
) {
815 AddDebugLogLineF(logAsio
, wxT("AcceptWith: nothing there"));
819 // return the socket we received
820 socket
.LinkSocketImpl(m_currentSocket
.release());
822 // check if we have another socket ready for reception
823 m_currentSocket
.reset(new CAsioSocketImpl(NULL
));
825 // async_accept does not work if server is non-blocking
826 // temporarily switch it to non-blocking
828 // we are set to non-blocking, so this returns right away
829 accept(m_currentSocket
->GetAsioSocket(), ec
);
832 if (ec
|| !m_currentSocket
->UpdateIP()) {
834 m_socketAvailable
= false;
835 // start getting another one
837 AddDebugLogLineF(logAsio
, wxT("AcceptWith: ok, getting another socket in background"));
839 // we got another socket right away
840 m_socketAvailable
= true; // it is already true, but this improves readability
841 AddDebugLogLineF(logAsio
, wxT("AcceptWith: ok, another socket is available"));
842 // aMule actually doesn't need a notification as it polls the listen socket.
843 // amuleweb does need it though
844 CoreNotify_ServerTCPAccept(m_libSocketServer
);
850 bool SocketAvailable() const { return m_socketAvailable
; }
856 m_currentSocket
.reset(new CAsioSocketImpl(NULL
));
857 async_accept(m_currentSocket
->GetAsioSocket(),
858 m_strand
.wrap(boost::bind(& CAsioSocketServerImpl::HandleAccept
, this, placeholders::error
)));
861 void HandleAccept(const error_code
& error
)
864 AddDebugLogLineC(logAsio
, CFormat(wxT("Error in HandleAccept: %s")) % error
.message());
866 if (m_currentSocket
->UpdateIP()) {
867 AddDebugLogLineN(logAsio
, CFormat(wxT("HandleAccept received a connection from %s:%d"))
868 % m_currentSocket
->GetIP() % m_currentSocket
->GetPort());
869 m_socketAvailable
= true;
870 CoreNotify_ServerTCPAccept(m_libSocketServer
);
873 AddDebugLogLineN(logAsio
, wxT("Error in HandleAccept: invalid socket"));
876 // We were not successful. Try again.
877 // Post the request to the event queue to make sure it doesn't get called immediately.
878 m_strand
.post(boost::bind(& CAsioSocketServerImpl::StartAccept
, this));
881 // The wrapper object
882 CLibSocketServer
* m_libSocketServer
;
885 // The last socket that connected to us
886 CScopedPtr
<CAsioSocketImpl
> m_currentSocket
;
887 // Is there a socket available?
888 bool m_socketAvailable
;
889 io_service::strand m_strand
; // handle synchronisation in io_service thread pool
893 CLibSocketServer::CLibSocketServer(const amuleIPV4Address
& adr
, int /* flags */)
895 m_aServer
= new CAsioSocketServerImpl(adr
, this);
899 CLibSocketServer::~CLibSocketServer()
905 // Accepts an incoming connection request, and creates a new CLibSocket object which represents the server-side of the connection.
906 // Only used in CamuleApp::ListenSocketHandler() and we don't get there.
907 CLibSocket
* CLibSocketServer::Accept(bool /* wait */)
914 // Accept an incoming connection using the specified socket object.
915 bool CLibSocketServer::AcceptWith(CLibSocket
& socket
, bool WXUNUSED_UNLESS_DEBUG(wait
) )
918 return m_aServer
->AcceptWith(socket
);
922 bool CLibSocketServer::IsOk() const
924 return m_aServer
->IsOk();
928 void CLibSocketServer::Close()
934 bool CLibSocketServer::SocketAvailable()
936 return m_aServer
->SocketAvailable();
941 * ASIO UDP socket implementation
944 class CAsioUDPSocketImpl
952 amuleIPV4Address ipadr
;
954 CUDPData(const void * src
, uint32 _size
, amuleIPV4Address adr
) :
955 size(_size
), ipadr(adr
)
957 buffer
= new char[size
];
958 memcpy(buffer
, src
, size
);
968 CAsioUDPSocketImpl(const amuleIPV4Address
&address
, int /* flags */, CLibUDPSocket
* libSocket
) :
969 m_libSocket(libSocket
),
970 m_strand(s_io_service
),
971 m_timer(s_io_service
),
976 m_readBuffer
= new char[CMuleUDPSocket::UDP_BUFFER_SIZE
];
981 ~CAsioUDPSocketImpl()
983 AddDebugLogLineF(logAsio
, wxT("UDP ~CAsioUDPSocketImpl"));
985 delete[] m_readBuffer
;
986 DeleteContents(m_receiveBuffers
);
989 void SetClientData(CMuleUDPSocket
* muleSocket
)
991 AddDebugLogLineF(logAsio
, wxT("UDP SetClientData"));
992 m_muleSocket
= muleSocket
;
995 uint32
RecvFrom(amuleIPV4Address
& addr
, void* buf
, uint32 nBytes
)
999 wxMutexLocker
lock(m_receiveBuffersLock
);
1000 if (m_receiveBuffers
.empty()) {
1001 AddDebugLogLineN(logAsio
, wxT("UDP RecvFromError no data"));
1004 recdata
= * m_receiveBuffers
.begin();
1005 m_receiveBuffers
.pop_front();
1007 uint32 read
= recdata
->size
;
1008 if (read
> nBytes
) {
1009 // should not happen
1010 AddDebugLogLineN(logAsio
, CFormat(wxT("UDP RecvFromError too much data %d")) % read
);
1013 memcpy(buf
, recdata
->buffer
, read
);
1014 addr
= recdata
->ipadr
;
1019 uint32
SendTo(const amuleIPV4Address
& addr
, const void* buf
, uint32 nBytes
)
1021 // Collect data, make a copy of the buffer's content
1022 CUDPData
* recdata
= new CUDPData(buf
, nBytes
, addr
);
1023 AddDebugLogLineF(logAsio
, CFormat(wxT("UDP SendTo %d to %s")) % nBytes
% addr
.IPAddress());
1024 m_strand
.dispatch(boost::bind(& CAsioUDPSocketImpl::DispatchSendTo
, this, recdata
));
1035 if (s_io_service
.stopped()) {
1038 m_strand
.dispatch(boost::bind(& CAsioUDPSocketImpl::DispatchClose
, this));
1044 AddDebugLogLineF(logAsio
, CFormat(wxT("Destroy() %p %p")) % m_libSocket
% this);
1046 if (s_io_service
.stopped()) {
1049 // Close prevents creation of any more callbacks, but does not clear any callbacks already
1050 // sitting in Asio's event queue (I have seen such a crash).
1051 // So create a delay timer so they can be called until core is notified.
1052 m_timer
.expires_from_now(boost::posix_time::seconds(1));
1053 m_timer
.async_wait(m_strand
.wrap(boost::bind(& CAsioUDPSocketImpl::HandleDestroy
, this)));
1060 // Dispatch handlers
1061 // Access to m_socket is all bundled in the thread running s_io_service to avoid
1062 // concurrent access to the socket from several threads.
1063 // So once things are running (after connect), all access goes through one of these handlers.
1065 void DispatchClose()
1068 m_socket
->close(ec
);
1070 AddDebugLogLineC(logAsio
, CFormat(wxT("UDP Close error %s")) % ec
.message());
1072 AddDebugLogLineF(logAsio
, wxT("UDP Closed"));
1076 void DispatchSendTo(CUDPData
* recdata
)
1078 ip::udp::endpoint
endpoint(recdata
->ipadr
.GetEndpoint().address(), recdata
->ipadr
.Service());
1080 AddDebugLogLineF(logAsio
, CFormat(wxT("UDP DispatchSendTo %d to %s:%d")) % recdata
->size
1081 % endpoint
.address().to_string() % endpoint
.port());
1082 m_socket
->async_send_to(buffer(recdata
->buffer
, recdata
->size
), endpoint
,
1083 m_strand
.wrap(boost::bind(& CAsioUDPSocketImpl::HandleSendTo
, this, placeholders::error
, placeholders::bytes_transferred
, recdata
)));
1087 // Completion handlers for async requests
1090 void HandleRead(const error_code
& ec
, size_t received
)
1093 AddDebugLogLineN(logAsio
, CFormat(wxT("UDP HandleReadError %s")) % ec
.message());
1094 } else if (received
== 0) {
1095 AddDebugLogLineF(logAsio
, wxT("UDP HandleReadError nothing available"));
1096 } else if (m_muleSocket
== NULL
) {
1097 AddDebugLogLineN(logAsio
, wxT("UDP HandleReadError no handler"));
1100 amuleIPV4Address ipadr
= amuleIPV4Address(CamuleIPV4Endpoint(m_receiveEndpoint
));
1101 AddDebugLogLineF(logAsio
, CFormat(wxT("UDP HandleRead %d %s:%d")) % received
% ipadr
.IPAddress() % ipadr
.Service());
1103 // create our read buffer
1104 CUDPData
* recdata
= new CUDPData(m_readBuffer
, received
, ipadr
);
1106 wxMutexLocker
lock(m_receiveBuffersLock
);
1107 m_receiveBuffers
.push_back(recdata
);
1109 CoreNotify_UDPSocketReceive(m_muleSocket
);
1111 StartBackgroundRead();
1114 void HandleSendTo(const error_code
& ec
, size_t sent
, CUDPData
* recdata
)
1117 AddDebugLogLineN(logAsio
, CFormat(wxT("UDP HandleSendToError %s")) % ec
.message());
1118 } else if (sent
!= recdata
->size
) {
1119 AddDebugLogLineN(logAsio
, CFormat(wxT("UDP HandleSendToError tosend: %d sent %d")) % recdata
->size
% sent
);
1121 if (m_muleSocket
== NULL
) {
1122 AddDebugLogLineN(logAsio
, wxT("UDP HandleSendToError no handler"));
1124 AddDebugLogLineF(logAsio
, CFormat(wxT("UDP HandleSendTo %d to %s")) % sent
% recdata
->ipadr
.IPAddress());
1125 CoreNotify_UDPSocketSend(m_muleSocket
);
1130 void HandleDestroy()
1132 AddDebugLogLineF(logAsio
, CFormat(wxT("HandleDestroy() %p %p")) % m_libSocket
% this);
1144 ip::udp::endpoint
endpoint(m_address
.GetEndpoint().address(), m_address
.Service());
1145 m_socket
= new ip::udp::socket(s_io_service
, endpoint
);
1146 AddDebugLogLineN(logAsio
, CFormat(wxT("Created UDP socket %s %d")) % m_address
.IPAddress() % m_address
.Service());
1147 StartBackgroundRead();
1148 } catch (const system_error
& err
) {
1149 AddLogLineC(CFormat(wxT("Error creating UDP socket %s %d : %s")) % m_address
.IPAddress() % m_address
.Service() % err
.code().message());
1155 void StartBackgroundRead()
1157 m_socket
->async_receive_from(buffer(m_readBuffer
, CMuleUDPSocket::UDP_BUFFER_SIZE
), m_receiveEndpoint
,
1158 m_strand
.wrap(boost::bind(& CAsioUDPSocketImpl::HandleRead
, this, placeholders::error
, placeholders::bytes_transferred
)));
1161 CLibUDPSocket
* m_libSocket
;
1162 ip::udp::socket
* m_socket
;
1163 CMuleUDPSocket
* m_muleSocket
;
1165 io_service::strand m_strand
; // handle synchronisation in io_service thread pool
1166 deadline_timer m_timer
;
1167 amuleIPV4Address m_address
;
1169 // One fix receive buffer
1170 char * m_readBuffer
;
1171 // and a list of dynamic buffers. UDP data may be coming in faster
1172 // than the main loop can handle it.
1173 std::list
<CUDPData
*> m_receiveBuffers
;
1174 wxMutex m_receiveBuffersLock
;
1176 // Address of last reception
1177 ip::udp::endpoint m_receiveEndpoint
;
1182 * Library UDP socket wrapper
1185 CLibUDPSocket::CLibUDPSocket(amuleIPV4Address
&address
, int flags
)
1187 m_aSocket
= new CAsioUDPSocketImpl(address
, flags
, this);
1191 CLibUDPSocket::~CLibUDPSocket()
1193 AddDebugLogLineF(logAsio
, CFormat(wxT("~CLibUDPSocket() %p %p")) % this % m_aSocket
);
1198 bool CLibUDPSocket::IsOk() const
1200 return m_aSocket
->IsOk();
1204 uint32
CLibUDPSocket::RecvFrom(amuleIPV4Address
& addr
, void* buf
, uint32 nBytes
)
1206 return m_aSocket
->RecvFrom(addr
, buf
, nBytes
);
1210 uint32
CLibUDPSocket::SendTo(const amuleIPV4Address
& addr
, const void* buf
, uint32 nBytes
)
1212 return m_aSocket
->SendTo(addr
, buf
, nBytes
);
1216 void CLibUDPSocket::SetClientData(CMuleUDPSocket
* muleSocket
)
1218 m_aSocket
->SetClientData(muleSocket
);
1222 int CLibUDPSocket::LastError() const
1228 void CLibUDPSocket::Close()
1234 void CLibUDPSocket::Destroy()
1236 m_aSocket
->Destroy();
1241 * CAsioService - ASIO event loop thread
1244 class CAsioServiceThread
: public wxThread
{
1246 CAsioServiceThread() : wxThread(wxTHREAD_JOINABLE
)
1248 static int count
= 0;
1249 m_threadNumber
= ++count
;
1256 AddLogLineNS(CFormat(_("Asio thread %d started")) % m_threadNumber
);
1257 io_service::work
worker(s_io_service
); // keep io_service running
1259 AddDebugLogLineN(logAsio
, CFormat(wxT("Asio thread %d stopped")) % m_threadNumber
);
1269 * The constructor starts the thread.
1271 CAsioService::CAsioService()
1273 m_threads
= new CAsioServiceThread
[m_numberOfThreads
];
1277 CAsioService::~CAsioService()
1282 void CAsioService::Stop()
1287 s_io_service
.stop();
1288 // Wait for threads to exit
1289 for (int i
= 0; i
< m_numberOfThreads
; i
++) {
1290 CAsioServiceThread
* t
= m_threads
+ i
;
1305 amuleIPV4Address::amuleIPV4Address()
1307 m_endpoint
= new CamuleIPV4Endpoint();
1310 amuleIPV4Address::amuleIPV4Address(const amuleIPV4Address
&a
)
1315 amuleIPV4Address::amuleIPV4Address(const CamuleIPV4Endpoint
&ep
)
1320 amuleIPV4Address::~amuleIPV4Address()
1325 amuleIPV4Address
& amuleIPV4Address::operator=(const amuleIPV4Address
&a
)
1327 m_endpoint
= new CamuleIPV4Endpoint(* a
.m_endpoint
);
1331 amuleIPV4Address
& amuleIPV4Address::operator=(const CamuleIPV4Endpoint
&ep
)
1333 m_endpoint
= new CamuleIPV4Endpoint(ep
);
1337 bool amuleIPV4Address::Hostname(const wxString
& name
)
1339 if (name
.IsEmpty()) {
1342 // This is usually just an IP.
1343 std::string
sname(unicode2char(name
));
1345 ip::address_v4 adr
= ip::address_v4::from_string(sname
, ec
);
1347 m_endpoint
->address(adr
);
1350 AddDebugLogLineN(logAsio
, CFormat(wxT("Hostname(\"%s\") failed, not an IP address %s")) % name
% ec
.message());
1352 // Try to resolve (sync). Normally not required. Unless you type in your hostname as "local IP address" or something.
1354 ip::tcp::resolver
res(s_io_service
);
1355 // We only want to get IPV4 addresses.
1356 ip::tcp::resolver::query
query(ip::tcp::v4(), sname
, "");
1357 ip::tcp::resolver::iterator endpoint_iterator
= res
.resolve(query
, ec2
);
1359 AddDebugLogLineN(logAsio
, CFormat(wxT("Hostname(\"%s\") resolve failed: %s")) % name
% ec2
.message());
1362 if (endpoint_iterator
== ip::tcp::resolver::iterator()) {
1363 AddDebugLogLineN(logAsio
, CFormat(wxT("Hostname(\"%s\") resolve failed: no address found")) % name
);
1366 m_endpoint
->address(endpoint_iterator
->endpoint().address());
1367 AddDebugLogLineN(logAsio
, CFormat(wxT("Hostname(\"%s\") resolved to %s")) % name
% IPAddress());
1371 bool amuleIPV4Address::Service(uint16 service
)
1376 m_endpoint
->port(service
);
1380 uint16
amuleIPV4Address::Service() const
1382 return m_endpoint
->port();
1385 bool amuleIPV4Address::IsLocalHost() const
1387 return m_endpoint
->address().is_loopback();
1390 wxString
amuleIPV4Address::IPAddress() const
1392 return CFormat(wxT("%s")) % m_endpoint
->address().to_string();
1395 // "Set address to any of the addresses of the current machine."
1396 // This just sets the address to 0.0.0.0 .
1397 // wx does the same.
1398 bool amuleIPV4Address::AnyAddress()
1400 m_endpoint
->address(ip::address_v4::any());
1401 AddDebugLogLineN(logAsio
, CFormat(wxT("AnyAddress: set to %s")) % IPAddress());
1405 const CamuleIPV4Endpoint
& amuleIPV4Address::GetEndpoint() const
1407 return * m_endpoint
;
1410 CamuleIPV4Endpoint
& amuleIPV4Address::GetEndpoint()
1412 return * m_endpoint
;
1417 // Notification stuff
1419 namespace MuleNotify
1422 void LibSocketConnect(CLibSocket
* socket
, int error
)
1424 if (socket
->IsDestroying()) {
1425 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketConnect Destroying %s %d")) % socket
->GetIP() % error
);
1426 } else if (socket
->GetProxyState()) {
1427 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketConnect Proxy %s %d")) % socket
->GetIP() % error
);
1428 socket
->OnProxyEvent(MULE_SOCKET_CONNECTION
);
1430 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketConnect %s %d")) %socket
->GetIP() % error
);
1431 socket
->OnConnect(error
);
1435 void LibSocketSend(CLibSocket
* socket
, int error
)
1437 if (socket
->IsDestroying()) {
1438 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketSend Destroying %s %d")) % socket
->GetIP() % error
);
1439 } else if (socket
->GetProxyState()) {
1440 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketSend Proxy %s %d")) % socket
->GetIP() % error
);
1441 socket
->OnProxyEvent(MULE_SOCKET_OUTPUT
);
1443 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketSend %s %d")) % socket
->GetIP() % error
);
1444 socket
->OnSend(error
);
1448 void LibSocketReceive(CLibSocket
* socket
, int error
)
1450 socket
->EventProcessed();
1451 if (socket
->IsDestroying()) {
1452 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketReceive Destroying %s %d")) % socket
->GetIP() % error
);
1453 } else if (socket
->GetProxyState()) {
1454 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketReceive Proxy %s %d")) % socket
->GetIP() % error
);
1455 socket
->OnProxyEvent(MULE_SOCKET_INPUT
);
1457 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketReceive %s %d")) % socket
->GetIP() % error
);
1458 socket
->OnReceive(error
);
1462 void LibSocketLost(CLibSocket
* socket
)
1464 if (socket
->IsDestroying()) {
1465 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketLost Destroying %s")) % socket
->GetIP());
1466 } else if (socket
->GetProxyState()) {
1467 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketLost Proxy %s")) % socket
->GetIP());
1468 socket
->OnProxyEvent(MULE_SOCKET_LOST
);
1470 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocketLost %s")) % socket
->GetIP());
1475 void LibSocketDestroy(CLibSocket
* socket
)
1477 AddDebugLogLineF(logAsio
, CFormat(wxT("LibSocket_Destroy %s")) % socket
->GetIP());
1481 void ProxySocketEvent(CLibSocket
* socket
, int evt
)
1483 AddDebugLogLineF(logAsio
, CFormat(wxT("ProxySocketEvent %s %d")) % socket
->GetIP() % evt
);
1484 socket
->OnProxyEvent(evt
);
1487 void ServerTCPAccept(CLibSocketServer
* socketServer
)
1489 AddDebugLogLineF(logAsio
, wxT("ServerTCP_Accept"));
1490 socketServer
->OnAccept();
1493 void UDPSocketSend(CMuleUDPSocket
* socket
)
1495 AddDebugLogLineF(logAsio
, wxT("UDPSocketSend"));
1499 void UDPSocketReceive(CMuleUDPSocket
* socket
)
1501 AddDebugLogLineF(logAsio
, wxT("UDPSocketReceive"));
1502 socket
->OnReceive(0);
1506 } // namespace MuleNotify
1509 // Initialize MuleBoostVersion
1511 wxString MuleBoostVersion
= CFormat(wxT("%d.%d")) % (BOOST_VERSION
/ 100000) % (BOOST_VERSION
/ 100 % 1000);