message passing

pch.h

@@ -94,8 +94,7 @@ namespace mongo {
 #include "string.h"
 #include "limits.h"
 
-///using namespace std;
-
+#include <boost/any.hpp>
 #include <boost/archive/iterators/base64_from_binary.hpp>
 #include <boost/archive/iterators/binary_from_base64.hpp>
 #include <boost/archive/iterators/transform_width.hpp>

util/concurrency/msg.h

@@ -18,39 +18,39 @@
 
 #pragma once
 
+#include <deque>
 #include "task.h"
 
 namespace mongo { 
 
     namespace task { 
 
-        template<class T>
+        /** See task::PortUnitTest in task.cpp for an example of usage.
+        */
         class Port : private Task { 
         protected:
-            /** implement a receiver of messages for the port. */
-            virtual void got(const T& msg) = 0;
-            virtual string name() = 0;
+            /** implement a receiver of messages for the port. 
+                @return false to stop the port / terminate the thread (i.e. you want to return true)
+            */
+            virtual bool got(const any& msg) = 0;
+            virtual string name() = 0; // names the thread
         public:
             /** send a message to the port */
-            void send(const T& msg);
+            void send(const any& msg);
 
             /** typical usage is: task::fork( foo.task() ); */
             shared_ptr<Task> taskPtr() { return shared_ptr<Task>(static_cast<Task*>(this)); }
 
             Port() { }
-            virtual ~Port() { 
-                cout << "TEMP PORT DONE " << endl;
-            }
+            virtual ~Port() { }
 
         private:
             void doWork();
-            mongo::mutex a,b;
+            deque<any> d;
+            boost::mutex m;
+            boost::condition c;
         };
 
-        template<class T>
-        inline void Port<T>::doWork() { 
-        }
-
     }
 
 }

util/concurrency/task.cpp

@@ -87,31 +87,57 @@ namespace mongo {
 
 #include "msg.h"
 
+/* class Task::Port */
+
 namespace mongo {
     namespace task {
 
-        /* tests for messaging - see msg.h */
+        void Port::send(const any& msg) { 
+            {
+                boost::mutex::scoped_lock lk(m);
+                d.push_back(msg);
+            }
+            c.notify_one();
+        }
 
-        /*
-        class JustTesting : public Port<int> {
+        void Port::doWork() { 
+            while( 1 ) { 
+                any a;
+                {
+                    boost::mutex::scoped_lock lk(m);
+                    while( d.empty() )
+                        c.wait(m);
+                    a = d.front();
+                    d.pop_front();
+                }
+                try {
+                    if( !got(a) )
+                        break;
+                } catch(std::exception& e) { 
+                    log() << "Port::doWork() exception " << e.what() << endl;
+                }
+            }
+        }
+        
+        class PortTest : public Port {
         protected:
             void got(const int& msg) { }
         public:
-            virtual string name() { return "ASD"; }
-            JustTesting() { }
+            virtual bool got(const any& msg) { 
+                assert( any_cast<int>(msg) <= 55 );
+                return any_cast<int>(msg) != 55;
+            }
+            virtual string name() { return "PortTest"; }
         };    
 
-        struct JTTest : public UnitTest {
+        struct PortUnitTest : public UnitTest {
             void run() { 
-                foo();
-                JustTesting *jt = new JustTesting();
-                shared_ptr<Task> tp = jt->taskPtr();
-                Task *t = tp.get();
+                PortTest *p = new PortTest();
+                shared_ptr<Task> tp = p->taskPtr();
                 fork( tp );
-                cout << "POKSDFFDSFDSFDSFDSFDS" << endl;
-
+                p->send(3);
+                p->send(55);
             } 
-        } juttt;
-        */
+        } portunittest;
     }
 }