4 files changed, 84 insertions, 16 deletions

diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp
index e0e95450..6ef24a7b 100644
--- a/src/stateless/cp/trex_stateless.cpp
+++ b/src/stateless/cp/trex_stateless.cpp
@@ -47,7 +47,7 @@ TrexStateless::TrexStateless(const TrexStatelessCfg &cfg) {
     m_port_count = cfg.m_port_count;
 
     for (int i = 0; i < m_port_count; i++) {
-        m_ports.push_back(new TrexStatelessPort(i));
+        m_ports.push_back(new TrexStatelessPort(i, cfg.m_platform_api));
     }
 
     m_platform_api = cfg.m_platform_api;
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index cbc5a328..13d0fc9f 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -52,9 +52,25 @@ using namespace std;
  * trex stateless port
  * 
  **************************/
-TrexStatelessPort::TrexStatelessPort(uint8_t port_id) : m_port_id(port_id) {
+TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api) {
+    std::vector<std::pair<uint8_t, uint8_t>> core_pair_list;
+
+    m_port_id = port_id;
+
     m_port_state = PORT_STATE_IDLE;
     clear_owner();
+
+    /* get the DP cores belonging to this port */
+    api->port_id_to_cores(m_port_id, core_pair_list);
+
+    for (auto core_pair : core_pair_list) {
+
+        /* send the core id */
+        m_cores_id_list.push_back(core_pair.first);
+    }
+
+    /* init the events DP DB */
+    m_dp_events.create(this);
 }
 
 
@@ -105,11 +121,16 @@ TrexStatelessPort::start_traffic(double mul, double duration) {
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj, duration);
+    m_event_id = m_dp_events.generate_event_id();
+    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_event_id, compiled_obj, duration);
+
+    change_state(PORT_STATE_TX);
 
     send_message_to_dp(start_msg);
 
-    change_state(PORT_STATE_TX);
+    /* mark that DP event of stoppped is possible */
+    m_dp_events.wait_for_event(TrexDpPortEvent::EVENT_STOP, m_event_id);
+    
 }
 
 /**
@@ -279,15 +300,32 @@ TrexStatelessPort::encode_stats(Json::Value &port) {
 void 
 TrexStatelessPort::send_message_to_dp(TrexStatelessCpToDpMsgBase *msg) {
 
-    std::vector<std::pair<uint8_t, uint8_t>> cores_id_list;
-
-    get_stateless_obj()->get_platform_api()->port_id_to_cores(m_port_id, cores_id_list);
-
-    for (auto core_pair : cores_id_list) {
+    for (auto core_id : m_cores_id_list) {
 
         /* send the message to the core */
-        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_pair.first);
+        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_id);
         ring->Enqueue((CGenNode *)msg->clone());
     }
 
 }
+
+/**
+ * when a DP (async) event occurs - handle it
+ * 
+ */
+void 
+TrexStatelessPort::on_dp_event_occured(TrexDpPortEvent::event_e event_type) {
+    switch (event_type) {
+
+    case TrexDpPortEvent::EVENT_STOP:
+        /* set a stop event */
+        change_state(PORT_STATE_STREAMS);
+        /* send a ZMQ event */
+        break;
+
+    default:
+        assert(0);
+
+    }
+    printf("hey");
+}
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index b533f793..da75284e 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -22,7 +22,9 @@ limitations under the License.
 #define __TREX_STATELESS_PORT_H__
 
 #include <trex_stream.h>
+#include <trex_dp_port_events.h>
 
+class TrexPlatformApi;
 class TrexStatelessCpToDpMsgBase;
 
 /**
@@ -31,6 +33,8 @@ class TrexStatelessCpToDpMsgBase;
  * @author imarom (31-Aug-15)
  */
 class TrexStatelessPort {
+    friend class TrexDpPortEvent;
+
 public:
 
     /**
@@ -54,7 +58,7 @@ public:
         RC_ERR_FAILED_TO_COMPILE_STREAMS
     };
 
-    TrexStatelessPort(uint8_t port_id);
+    TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api);
     
     /**
      * acquire port
@@ -199,6 +203,10 @@ public:
         m_stream_table.get_object_list(object_list);
     }
 
+    TrexDpPortEvents & get_dp_events() {
+        return m_dp_events;
+    }
+
 private:
 
   
@@ -224,6 +232,10 @@ private:
     }
 
 
+    const std::vector<int> get_core_id_list () {
+        return m_cores_id_list;
+    }
+
     bool verify_state(int state, bool should_throw = true) const;
 
     void change_state(port_state_e new_state);
@@ -232,11 +244,25 @@ private:
 
     void send_message_to_dp(TrexStatelessCpToDpMsgBase *msg);
 
-    TrexStreamTable  m_stream_table;
-    uint8_t          m_port_id;
-    port_state_e     m_port_state;
-    std::string      m_owner;
-    std::string      m_owner_handler;
+    /**
+     * triggered when event occurs
+     * 
+     */
+    void on_dp_event_occured(TrexDpPortEvent::event_e event_type);
+
+
+    TrexStreamTable    m_stream_table;
+    uint8_t            m_port_id;
+    port_state_e       m_port_state;
+    std::string        m_owner;
+    std::string        m_owner_handler;
+
+    /* holds the DP cores associated with this port */
+    //std::vector<std::pair<uint8_t, uint8_t>> m_cores_id_list;
+    std::vector<int> m_cores_id_list;
+
+    TrexDpPortEvents   m_dp_events;
+    int                m_event_id;
 };
 
 #endif /* __TREX_STATELESS_PORT_H__ */
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index 42cfc5b8..c80dddef 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -60,6 +60,10 @@ public:
         return (m_mul);
     }
 
+    uint8_t get_port_id() {
+        return m_port_id;
+    }
+
 private:
     void add_compiled_stream(TrexStream * stream);
     std::vector<obj_st> m_objs;