Changeset 10851 for trunk

Timestamp:

05/27/11 11:40:24 (13 years ago)

Author:

tbretz

Message:

Added support for switching between command socket and data sockets.

File:

• trunk/FACT++/src/fad.cc (modified) (9 diffs)

trunk/FACT++/src/fad.cc

@@ -40 +40 @@
     }
 
-    void AsyncWrite(const ba::const_buffers_1 &buffers)
-    {
-        ba::async_write(*this, buffers,
+    void AsyncWrite(ba::ip::tcp::socket *socket, const ba::const_buffers_1 &buffers)
+    {
+        ba::async_write(*socket, buffers,
                         boost::bind(&tcp_connection::HandleSentData, shared_from_this(),
                                     dummy::error, dummy::bytes_transferred));
@@ -76 +76 @@
 
     bool fTriggerEnabled;
+    bool fCommandSocket;
+
+    int fSocket;
 
     void SendData()
@@ -108 +111 @@
         fBuffer.insert(fBuffer.begin(), h.begin(), h.end());
 
-        AsyncWrite(ba::buffer(ba::const_buffer(fBuffer.data(), fBuffer.size()*2)));
+        if (fCommandSocket)
+            AsyncWrite(this, ba::buffer(ba::const_buffer(fBuffer.data(), fBuffer.size()*2)));
+        else
+        {
+            fSocket++;
+            fSocket %= fSockets.size();
+
+            AsyncWrite(fSockets[fSocket].get(), ba::buffer(ba::const_buffer(fBuffer.data(), fBuffer.size()*2)));
+        }
     }
 
@@ -186 +197 @@
             case kCmdRun+0x100:
                 cout << "-> Run" << endl;
+                break;
+
+            case kCmdSocket:
+            case kCmdSocket+0x100:
+                cout << "-> Socket" << endl;
+                fCommandSocket = fBufCommand[0]==kCmdSocket;
                 break;
 
@@ -267 +284 @@
         // Ownership of buffer must be valid until Handler is called.
 
+        fTriggerEnabled=false;
+        fCommandSocket=true;
+
         fHeader.fStartDelimiter = FAD::kDelimiterStart;
         fHeader.fVersion = 0x104;
@@ -294 +314 @@
 
     }
+
+    vector<boost::shared_ptr<ba::ip::tcp::socket>> fSockets;
+
+    ~tcp_connection()
+    {
+        fSockets.clear();
+    }
+
+    void handle_accept(boost::shared_ptr<ba::ip::tcp::socket> socket, int port, const boost::system::error_code&/* error*/)
+    {
+        cout << "Added one socket " << socket->remote_endpoint().address().to_v4().to_string();
+        cout << ":"<< port << endl;
+        fSockets.push_back(socket);
+    }
 };
 
 
-class tcp_server : public tcp::acceptor
+class tcp_server
 {
+    tcp::acceptor acc0;
+    tcp::acceptor acc1;
+    tcp::acceptor acc2;
+    tcp::acceptor acc3;
+    tcp::acceptor acc4;
+    tcp::acceptor acc5;
+    tcp::acceptor acc6;
+    tcp::acceptor acc7;
+
+    int fPort;
+
 public:
     tcp_server(ba::io_service& ioservice, int port) :
-        tcp::acceptor(ioservice, tcp::endpoint(tcp::v4(), port))
-
+        acc0(ioservice, tcp::endpoint(tcp::v4(), port)),
+        acc1(ioservice, tcp::endpoint(tcp::v4(), port+1)),
+        acc2(ioservice, tcp::endpoint(tcp::v4(), port+2)),
+        acc3(ioservice, tcp::endpoint(tcp::v4(), port+3)),
+        acc4(ioservice, tcp::endpoint(tcp::v4(), port+4)),
+        acc5(ioservice, tcp::endpoint(tcp::v4(), port+5)),
+        acc6(ioservice, tcp::endpoint(tcp::v4(), port+6)),
+        acc7(ioservice, tcp::endpoint(tcp::v4(), port+7)),
+        fPort(port)
     {
         // We could start listening for more than one connection
@@ -313 +365 @@
 
 private:
+    void start_accept(tcp_connection::shared_ptr dest, tcp::acceptor &acc)
+    {
+        boost::shared_ptr<ba::ip::tcp::socket> connection =
+            boost::shared_ptr<ba::ip::tcp::socket>(new ba::ip::tcp::socket(acc.io_service()));
+        acc.async_accept(*connection,
+                          boost::bind(&tcp_connection::handle_accept,
+                                      dest, connection,
+                                      acc.local_endpoint().port(),
+                                      ba::placeholders::error));
+    }
+
     void start_accept()
     {
         cout << "Start accept..." << flush;
-        tcp_connection::shared_ptr new_connection = tcp_connection::create(/*acceptor_.*/io_service());
+        tcp_connection::shared_ptr new_connection = tcp_connection::create(/*acceptor_.*/acc0.io_service());
 
         // This will accept a connection without blocking
-        async_accept(*new_connection,
-                     boost::bind(&tcp_server::handle_accept,
-                                 this,
-                                 new_connection,
-                                 ba::placeholders::error));
-
+        acc0.async_accept(*new_connection,
+                          boost::bind(&tcp_server::handle_accept,
+                                      this,
+                                      new_connection,
+                                      ba::placeholders::error));
+
+        start_accept(new_connection, acc1);
+        start_accept(new_connection, acc2);
+        start_accept(new_connection, acc3);
+        start_accept(new_connection, acc4);
+        start_accept(new_connection, acc5);
+        start_accept(new_connection, acc6);
+        start_accept(new_connection, acc7);
 
         cout << "start-done." << endl;
@@ -358 +428 @@
         tcp_server server(io_service, port);
 
-        tcp::acceptor acc1(io_service, tcp::endpoint(tcp::v4(), port+1));
-        tcp::acceptor acc2(io_service, tcp::endpoint(tcp::v4(), port+2));
-        tcp::acceptor acc3(io_service, tcp::endpoint(tcp::v4(), port+3));
-        tcp::acceptor acc4(io_service, tcp::endpoint(tcp::v4(), port+4));
-        tcp::acceptor acc5(io_service, tcp::endpoint(tcp::v4(), port+5));
-        tcp::acceptor acc6(io_service, tcp::endpoint(tcp::v4(), port+6));
-        tcp::acceptor acc7(io_service, tcp::endpoint(tcp::v4(), port+7));
-
         //  ba::add_service(io_service, &server);
         //  server.add_service(...);
-        cout << "Run..." << flush;
+        //cout << "Run..." << flush;
 
         // Calling run() from a single thread ensures no concurrent access
@@ -374 +436 @@
         io_service.run();
 
-        cout << "end." << endl;
+        //cout << "end." << endl;
     }
     catch (std::exception& e)