diff options
| -rw-r--r-- | anyrtpproxy/Makefile | 4 | ||||
| -rw-r--r-- | anyrtpproxy/anyrtpproxy.cpp | 77 | ||||
| -rw-r--r-- | anyrtpproxy/options.h | 12 |
3 files changed, 90 insertions, 3 deletions
diff --git a/anyrtpproxy/Makefile b/anyrtpproxy/Makefile index f4486b1..354099b 100644 --- a/anyrtpproxy/Makefile +++ b/anyrtpproxy/Makefile @@ -1,7 +1,11 @@ C = gcc CFLAGS = -g -Wall +CFLAGS += -DSOCKETS_NAMESPACE=sockets +CFLAGS += -DSOCKETS_NAMESPACE_STR='"sockets"' C++ = g++ CCFLAGS = -g -Wall +CCFLAGS += -DSOCKETS_NAMESPACE=sockets +CCFLAGS += -DSOCKETS_NAMESPACE_STR='"sockets"' LD = g++ LDFLAGS = -g -Wall -O2 -lpthread diff --git a/anyrtpproxy/anyrtpproxy.cpp b/anyrtpproxy/anyrtpproxy.cpp index 3b0c488..da03ff4 100644 --- a/anyrtpproxy/anyrtpproxy.cpp +++ b/anyrtpproxy/anyrtpproxy.cpp @@ -10,6 +10,17 @@ #include "../signalController.h" #include "../PracticalSocket.h" #include "../buffer.h" +#include "../connectionList.h" +#include "../rtpSessionTable.h" +#include "../syncCommand.h" +#include "../syncQueue.h" +#include "../syncSocketHandler.h" +#include "../syncListenSocket.h" + +#include "../syncSocket.h" +#include "../syncClientSocket.h" +#include "../threadUtils.hpp" + #include "options.h" #include <map> @@ -17,6 +28,18 @@ #define MAX_PACKET_SIZE 1500 + +class ThreadParam +{ +public: + ThreadParam(SyncQueue & queue_,OptionConnectTo & connto_) + : queue(queue_),connto(connto_) + {}; + SyncQueue & queue; + OptionConnectTo & connto; +}; + + class ControlHost : public Host { public: @@ -33,7 +56,7 @@ public: private: - Mutex mutex; + ::Mutex mutex; std::map<ControlHost, std::pair<UDPSocket*, pthread_t> > control_hosts_; }; @@ -162,6 +185,40 @@ void daemonize() umask(027); } +void* syncConnector(void* p ) +{ + ThreadParam* param = reinterpret_cast<ThreadParam*>(p); + + SocketHandler h; + ConnectionList cl; + SyncClientSocket sock(h,cl); + sock.Open( param->connto.host, param->connto.port); + h.Add(&sock); + while (h.GetCount()) + { + h.Select(); + } + pthread_exit(NULL); +} + +void* syncListener(void* p ) +{ + ThreadParam* param = reinterpret_cast<ThreadParam*>(p); + ConnectionList cl; + + SyncSocketHandler h(param->queue); + SyncListenSocket<SyncSocket,ConnectionList> l(h,cl); + + if (l.Bind(gOpt.getLocalSyncPort())) + pthread_exit(NULL); + + Utility::ResolveLocal(); // resolve local hostname + h.Add(&l); + h.Select(1,0); + while (1) { + h.Select(1,0); + } +} int main(int argc, char* argv[]) { @@ -186,6 +243,24 @@ int main(int argc, char* argv[]) pthread_t senderThread; pthread_create(&senderThread, NULL, sender, NULL); pthread_detach(senderThread); + pthread_t syncListenerThread; + + SyncQueue queue; +// Example +// gRtpSessionTable.addSession(std::string("callid"),RtpSession()); +// SyncCommand sc (std::string("callid")); +// queue.push(sc); + ThreadParam p( queue,*(new OptionConnectTo())) + if ( gOpt.getLocalSyncPort()) + pthread_create(&syncListenerThread, NULL, syncListener, &p); + + std::list<pthread_t> connectThreads; + for(ConnectToList::iterator it = connect_to.begin() ;it != connect_to.end(); ++it) + { + connectThreads.push_back(pthread_t()); + ThreadParam * point = new ThreadParam(dev, *src, cl, queue,*it); + pthread_create(& connectThreads.back(), NULL, syncConnector, point); + } int ret = sig.run(); return ret; diff --git a/anyrtpproxy/options.h b/anyrtpproxy/options.h index c75da6a..25004a4 100644 --- a/anyrtpproxy/options.h +++ b/anyrtpproxy/options.h @@ -35,6 +35,14 @@ #include <list> #include <sstream> +typedef struct OptionConnectTo +{ + std::string host; + uint16_t port; +}; + +typedef std::list<OptionConnectTo> ConnectToList; + class Host { public: @@ -84,7 +92,7 @@ private: bool sanityCheck(); static Options* inst; - static Mutex instMutex; + static ::Mutex instMutex; class instanceCleaner { public: ~instanceCleaner() { if(Options::inst != 0) @@ -93,7 +101,7 @@ private: }; friend class instanceCleaner; - Mutex mutex; + ::Mutex mutex; std::string progname_; bool chroot_; |