9 files changed, 427 insertions, 170 deletions

diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index ae87d749..a2d4c284 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -25,11 +25,13 @@ limitations under the License.
 #include <trex_stateless_port.h>
 #include <trex_rpc_cmds_table.h>
 
+#include <internal_api/trex_platform_api.h>
+
 #include <fstream>
 #include <iostream>
 #include <unistd.h>
 
-#ifndef TREX_RPC_MOCK_SERVER
+#ifdef RTE_DPDK
     #include <../linux_dpdk/version.h>
 #endif
 
@@ -41,7 +43,7 @@ using namespace std;
 trex_rpc_cmd_rc_e 
 TrexRpcCmdPing::_run(const Json::Value &params, Json::Value &result) {
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
     return (TREX_RPC_CMD_OK);
 }
 
@@ -73,7 +75,7 @@ TrexRpcCmdGetVersion::_run(const Json::Value &params, Json::Value &result) {
 
     Json::Value &section = result["result"];
 
-    #ifndef TREX_RPC_MOCK_SERVER
+    #ifdef RTE_DPDK
 
     section["version"]       = VERSION_BUILD_NUM;
     section["build_date"]    = get_build_date();
@@ -145,7 +147,7 @@ trex_rpc_cmd_rc_e
 TrexRpcCmdGetSysInfo::_run(const Json::Value &params, Json::Value &result) {
     string hostname;
 
-    TrexStateless & instance = TrexStateless::get_instance();
+    TrexStateless * main = get_stateless_obj();
 
     Json::Value &section = result["result"];
 
@@ -155,30 +157,46 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value &params, Json::Value &result) {
     section["uptime"] = TrexRpcServer::get_server_uptime();
 
     /* FIXME: core count */
-    section["dp_core_count"] = instance.get_dp_core_count();
+    section["dp_core_count"] = main->get_dp_core_count();
     section["core_type"] = get_cpu_model();
 
     /* ports */
    
 
-    section["port_count"] = instance.get_port_count();
+    section["port_count"] = main->get_port_count();
 
     section["ports"] = Json::arrayValue;
 
-    for (int i = 0; i < instance.get_port_count(); i++) {
+    for (int i = 0; i < main->get_port_count(); i++) {
         string driver;
-        string speed;
+        TrexPlatformApi::driver_speed_e speed;
 
-        TrexStatelessPort *port = instance.get_port_by_id(i);
+        TrexStatelessPort *port = main->get_port_by_id(i);
         port->get_properties(driver, speed);
 
         section["ports"][i]["index"]   = i;
+
         section["ports"][i]["driver"]  = driver;
-        section["ports"][i]["speed"]   = speed;
 
-        section["ports"][i]["owner"] = port->get_owner();
+        switch (speed) {
+        case TrexPlatformApi::SPEED_1G:
+            section["ports"][i]["speed"]   = 1;
+            break;
+
+        case TrexPlatformApi::SPEED_10G:
+            section["ports"][i]["speed"]   = 10;
+            break;
+
+        case TrexPlatformApi::SPEED_40G:
+            section["ports"][i]["speed"]   = 40;
+            break;
+
+        default:
+            /* unknown value */
+            section["ports"][i]["speed"]   = 0;
+            break;
+        }
 
-        section["ports"][i]["status"] = port->get_state_as_string();
 
     }
 
@@ -201,8 +219,8 @@ TrexRpcCmdGetOwner::_run(const Json::Value &params, Json::Value &result) {
 
     uint8_t port_id = parse_port(params, result);
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
-    section["owner"] = port->get_owner();
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+    section["owner"] = port->get_owner().get_name();
 
     return (TREX_RPC_CMD_OK);
 }
@@ -216,19 +234,19 @@ TrexRpcCmdAcquire::_run(const Json::Value &params, Json::Value &result) {
 
     uint8_t port_id = parse_port(params, result);
 
-    const string &new_owner = parse_string(params, "user", result);
+    const string  &new_owner  = parse_string(params, "user", result);
     bool force = parse_bool(params, "force", result);
 
     /* if not free and not you and not force - fail */
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if ( (!port->is_free_to_aquire()) && (port->get_owner() != new_owner) && (!force)) {
-        generate_execute_err(result, "port is already taken by '" + port->get_owner() + "'");
+    try {
+        port->acquire(new_owner, force);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
     }
 
-    port->set_owner(new_owner);
-
-    result["result"] = port->get_owner_handler();
+    result["result"] = port->get_owner().get_handler();
 
     return (TREX_RPC_CMD_OK);
 }
@@ -242,15 +260,15 @@ TrexRpcCmdRelease::_run(const Json::Value &params, Json::Value &result) {
 
     uint8_t port_id = parse_port(params, result);
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if (port->get_state() == TrexStatelessPort::PORT_STATE_TRANSMITTING) {
-        generate_execute_err(result, "cannot release a port during transmission");
+    try {
+        port->release();
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
     }
 
-    port->clear_owner();
-
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -264,15 +282,36 @@ TrexRpcCmdGetPortStats::_run(const Json::Value &params, Json::Value &result) {
 
     uint8_t port_id = parse_port(params, result);
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if (port->get_state() == TrexStatelessPort::PORT_STATE_DOWN) {
-        generate_execute_err(result, "cannot get stats - port is down");
+    try {
+        port->encode_stats(result["result"]);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
     }
 
-    result["result"]["status"] = port->get_state_as_string();
+    return (TREX_RPC_CMD_OK);
+}
+
+/**
+ * fetch the port status
+ * 
+ * @author imarom (09-Dec-15)
+ * 
+ * @param params 
+ * @param result 
+ * 
+ * @return trex_rpc_cmd_rc_e 
+ */
+trex_rpc_cmd_rc_e
+TrexRpcCmdGetPortStatus::_run(const Json::Value &params, Json::Value &result) {
+    uint8_t port_id = parse_port(params, result);
+
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    result["result"]["owner"]  = (port->get_owner().is_free() ? "" : port->get_owner().get_name());
+    result["result"]["state"] = port->get_state_as_string();
 
-    port->encode_stats(result["result"]);
 
     return (TREX_RPC_CMD_OK);
 }
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index 20107411..fa3d96b2 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -23,6 +23,7 @@ limitations under the License.
 #include <trex_stream.h>
 #include <trex_stateless.h>
 #include <trex_stateless_port.h>
+#include <trex_streams_compiler.h>
 
 #include <iostream>
 
@@ -52,7 +53,8 @@ static uint64_t str2num(const string &str) {
 trex_rpc_cmd_rc_e
 TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
 
-    uint8_t  port_id    = parse_int(params, "port_id", result);
+    uint8_t port_id = parse_port(params, result);
+
     uint32_t stream_id  = parse_int(params, "stream_id", result);
 
     const Json::Value &section = parse_object(params, "stream", result);
@@ -114,10 +116,15 @@ TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
     /* make sure this is a valid stream to add */
     validate_stream(stream, result);
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(stream->m_port_id);
-    port->get_stream_table()->add_stream(stream);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(stream->m_port_id);
 
-    result["result"] = "ACK";
+    try {
+        port->add_stream(stream);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -127,7 +134,7 @@ TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
 TrexStream *
 TrexRpcCmdAddStream::allocate_new_stream(const Json::Value &section, uint8_t port_id, uint32_t stream_id, Json::Value &result) {
 
-    TrexStream *stream;
+    TrexStream *stream = NULL;
 
     const Json::Value &mode = parse_object(section, "mode", result);
     std::string type = parse_string(mode, "type", result);
@@ -135,14 +142,22 @@ TrexRpcCmdAddStream::allocate_new_stream(const Json::Value &section, uint8_t por
     if (type == "continuous") {
 
         double pps = parse_double(mode, "pps", result);
-        stream = new TrexStreamContinuous(port_id, stream_id, pps);
+        stream = new TrexStream( TrexStream::stCONTINUOUS, port_id, stream_id);
+        stream->set_pps(pps);
+
+        if (stream->m_next_stream_id != -1) {
+            generate_parse_err(result, "continious stream cannot provide next stream id - only -1 is valid");
+        }
 
     } else if (type == "single_burst") {
 
         uint32_t total_pkts      = parse_int(mode, "total_pkts", result);
         double pps               = parse_double(mode, "pps", result);
 
-        stream = new TrexStreamBurst(port_id, stream_id, total_pkts, pps);
+        stream = new TrexStream(TrexStream::stSINGLE_BURST,port_id, stream_id);
+        stream->set_pps(pps);
+        stream->set_single_burst(total_pkts);
+
 
     } else if (type == "multi_burst") {
 
@@ -151,8 +166,10 @@ TrexRpcCmdAddStream::allocate_new_stream(const Json::Value &section, uint8_t por
         uint32_t  num_bursts       = parse_int(mode, "number_of_bursts", result);
         uint32_t  pkts_per_burst   = parse_int(mode, "pkts_per_burst", result);
 
-        stream = new TrexStreamMultiBurst(port_id, stream_id, pkts_per_burst, pps, num_bursts, ibg_usec);
-        
+        stream = new TrexStream(TrexStream::stMULTI_BURST,port_id, stream_id );
+        stream->set_pps(pps);
+        stream->set_multi_burst(pkts_per_burst,num_bursts,ibg_usec);
+
 
     } else {
         generate_parse_err(result, "bad stream type provided: '" + type + "'");
@@ -200,9 +217,9 @@ TrexRpcCmdAddStream::parse_vm_instr_flow_var(const Json::Value &inst, TrexStream
     std::string  min_value_str     = parse_string(inst, "min_value", result);
     std::string  max_value_str     = parse_string(inst, "max_value", result);
 
-    uint64_t init_value;
-    uint64_t min_value;
-    uint64_t max_value;
+    uint64_t init_value = 0;
+    uint64_t min_value = 0;
+    uint64_t max_value = 0;
 
     try {
         init_value = str2num(init_value_str);
@@ -281,19 +298,11 @@ TrexRpcCmdAddStream::validate_stream(const TrexStream *stream, Json::Value &resu
         generate_execute_err(result, ss.str()); 
     }
 
-    /* port id should be between 0 and count - 1 */
-    if (stream->m_port_id >= TrexStateless::get_instance().get_port_count()) {
-        std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        delete stream;
-        generate_execute_err(result, ss.str());
-    }
-
-     /* add the stream to the port's stream table */
-    TrexStatelessPort * port = TrexStateless::get_instance().get_port_by_id(stream->m_port_id);
+    /* add the stream to the port's stream table */
+    TrexStatelessPort * port = get_stateless_obj()->get_port_by_id(stream->m_port_id);
 
     /* does such a stream exists ? */
-    if (port->get_stream_table()->get_stream_by_id(stream->m_stream_id)) {
+    if (port->get_stream_by_id(stream->m_stream_id)) {
         std::stringstream ss;
         ss << "stream " << stream->m_stream_id << " already exists";
         delete stream;
@@ -308,18 +317,12 @@ TrexRpcCmdAddStream::validate_stream(const TrexStream *stream, Json::Value &resu
  **************************/
 trex_rpc_cmd_rc_e
 TrexRpcCmdRemoveStream::_run(const Json::Value &params, Json::Value &result) {
-    uint8_t  port_id = parse_byte(params, "port_id", result);
-    uint32_t stream_id = parse_int(params, "stream_id", result);
 
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
-        std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        generate_execute_err(result, ss.str());
-    }
-
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
-    TrexStream *stream = port->get_stream_table()->get_stream_by_id(stream_id);
+    uint32_t stream_id = parse_int(params, "stream_id", result);
+    TrexStream *stream = port->get_stream_by_id(stream_id);
 
     if (!stream) {
         std::stringstream ss;
@@ -327,10 +330,15 @@ TrexRpcCmdRemoveStream::_run(const Json::Value &params, Json::Value &result) {
         generate_execute_err(result, ss.str());
     }
 
-    port->get_stream_table()->remove_stream(stream);
+    try {
+        port->remove_stream(stream);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+    
     delete stream;
 
-    result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
     return (TREX_RPC_CMD_OK);
 }
@@ -342,20 +350,20 @@ TrexRpcCmdRemoveStream::_run(const Json::Value &params, Json::Value &result) {
  **************************/
 trex_rpc_cmd_rc_e
 TrexRpcCmdRemoveAllStreams::_run(const Json::Value &params, Json::Value &result) {
-    uint8_t  port_id = parse_byte(params, "port_id", result);
 
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
-        std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        generate_execute_err(result, ss.str());
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    try {
+        port->remove_and_delete_all_streams();
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
     }
 
-       TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
-       port->get_stream_table()->remove_and_delete_all_streams();
 
-       result["result"] = "ACK";
+    result["result"] = Json::objectValue;
 
-       return (TREX_RPC_CMD_OK);
+    return (TREX_RPC_CMD_OK);
 }
 
 /***************************
@@ -367,27 +375,20 @@ trex_rpc_cmd_rc_e
 TrexRpcCmdGetStreamList::_run(const Json::Value &params, Json::Value &result) {
     std::vector<uint32_t> stream_list;
 
-    uint8_t  port_id = parse_byte(params, "port_id", result);
-
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
-        std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        generate_execute_err(result, ss.str());
-    }
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-       TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    port->get_id_list(stream_list);
 
-       port->get_stream_table()->get_stream_list(stream_list);
+    Json::Value json_list = Json::arrayValue;
 
-       Json::Value json_list = Json::arrayValue;
-
-       for (auto stream_id : stream_list) {
-           json_list.append(stream_id);
-       }
+    for (auto stream_id : stream_list) {
+        json_list.append(stream_id);
+    }
 
-       result["result"] = json_list;
+    result["result"] = json_list;
 
-       return (TREX_RPC_CMD_OK);
+    return (TREX_RPC_CMD_OK);
 }
 
 /***************************
@@ -397,19 +398,14 @@ TrexRpcCmdGetStreamList::_run(const Json::Value &params, Json::Value &result) {
  **************************/
 trex_rpc_cmd_rc_e
 TrexRpcCmdGetStream::_run(const Json::Value &params, Json::Value &result) {
-    uint8_t  port_id = parse_byte(params, "port_id", result);
 
-    uint32_t stream_id = parse_int(params, "stream_id", result);
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
-        std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        generate_execute_err(result, ss.str());
-    }
-
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    bool     get_pkt   = parse_bool(params, "get_pkt", result);
+    uint32_t stream_id = parse_int(params, "stream_id", result);
 
-    TrexStream *stream = port->get_stream_table()->get_stream_by_id(stream_id);
+    TrexStream *stream = port->get_stream_by_id(stream_id);
 
     if (!stream) {
         std::stringstream ss;
@@ -418,7 +414,12 @@ TrexRpcCmdGetStream::_run(const Json::Value &params, Json::Value &result) {
     }
 
     /* return the stored stream json (instead of decoding it all over again) */
-    result["result"]["stream"] = stream->get_stream_json();
+    Json::Value j = stream->get_stream_json();
+    if (!get_pkt) {
+        j.removeMember("packet");
+    }
+
+    result["result"]["stream"] = j;
 
     return (TREX_RPC_CMD_OK);
 
@@ -431,59 +432,216 @@ TrexRpcCmdGetStream::_run(const Json::Value &params, Json::Value &result) {
 trex_rpc_cmd_rc_e
 TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
 
-    uint8_t port_id = parse_byte(params, "port_id", result);
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
-        std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        generate_execute_err(result, ss.str());
-    }
+    double duration  = parse_double(params, "duration", result);
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    /* multiplier */
+    const Json::Value &mul_obj  = parse_object(params, "mul", result);
 
-    TrexStatelessPort::rc_e rc = port->start_traffic();
+    std::string type   = parse_choice(mul_obj, "type", TrexPortMultiplier::g_types, result);
+    std::string op     = parse_string(mul_obj, "op", result);
+    double      value  = parse_double(mul_obj, "value", result);
 
-    if (rc == TrexStatelessPort::RC_OK) {
-        result["result"] = "ACK";
-    } else {
-        std::stringstream ss;
-        switch (rc) {
-        case TrexStatelessPort::RC_ERR_BAD_STATE_FOR_OP:
-            ss << "bad state for operations: port is either transmitting traffic or down";
-            break;
-        case TrexStatelessPort::RC_ERR_NO_STREAMS:
-            ss << "no active streams on that port";
-            break;
-        default:
-            ss << "failed to start traffic";
-            break;
-        }
+    if (op != "abs") {
+        generate_parse_err(result, "start message can only specify absolute speed rate");
+    }
 
-        generate_execute_err(result, ss.str());
+    TrexPortMultiplier mul(type, op, value);
+
+    try {
+        port->start_traffic(mul, duration);
+
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
     }
 
+    result["result"]["multiplier"] = port->get_multiplier();
+
     return (TREX_RPC_CMD_OK);
 }
 
 /***************************
- * start traffic on port
+ * stop traffic on port
  * 
  **************************/
 trex_rpc_cmd_rc_e
 TrexRpcCmdStopTraffic::_run(const Json::Value &params, Json::Value &result) {
-    uint8_t  port_id = parse_byte(params, "port_id", result);
 
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    try {
+        port->stop_traffic();
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    result["result"] = Json::objectValue;
+
+    return (TREX_RPC_CMD_OK);
+}
+
+/***************************
+ * get all streams
+ * 
+ **************************/
+trex_rpc_cmd_rc_e
+TrexRpcCmdGetAllStreams::_run(const Json::Value &params, Json::Value &result) {
+    
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    bool    get_pkt = parse_bool(params, "get_pkt", result);
+
+    std::vector <TrexStream *> streams;
+    port->get_object_list(streams);
+
+    Json::Value streams_json = Json::objectValue;
+    for (auto stream : streams) {
+
+        Json::Value j = stream->get_stream_json();
+
+        /* should we include the packet as well ? */
+        if (!get_pkt) {
+            j.removeMember("packet");
+        }
+
         std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
-        generate_execute_err(result, ss.str());
+        ss << stream->m_stream_id;
+
+        streams_json[ss.str()] = j;
     }
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    result["result"]["streams"] = streams_json;
+
+    return (TREX_RPC_CMD_OK);
+}
+
+/***************************
+ * pause traffic
+ * 
+ **************************/
+trex_rpc_cmd_rc_e
+TrexRpcCmdPauseTraffic::_run(const Json::Value &params, Json::Value &result) {
+
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+     try {
+        port->pause_traffic();
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    result["result"] = Json::objectValue;
+
+    return (TREX_RPC_CMD_OK);
+}
+
+/***************************
+ * resume traffic
+ * 
+ **************************/
+trex_rpc_cmd_rc_e
+TrexRpcCmdResumeTraffic::_run(const Json::Value &params, Json::Value &result) {
+
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+     try {
+        port->resume_traffic();
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    result["result"] = Json::objectValue;
+
+    return (TREX_RPC_CMD_OK);
+}
+
+/***************************
+ * update traffic
+ * 
+ **************************/
+trex_rpc_cmd_rc_e
+TrexRpcCmdUpdateTraffic::_run(const Json::Value &params, Json::Value &result) {
+
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    /* multiplier */
+
+    const Json::Value &mul_obj  = parse_object(params, "mul", result);
+
+    std::string type   = parse_choice(mul_obj, "type", TrexPortMultiplier::g_types, result);
+    std::string op     = parse_choice(mul_obj, "op", TrexPortMultiplier::g_ops, result);
+    double      value  = parse_double(mul_obj, "value", result);
+
+    TrexPortMultiplier mul(type, op, value);
 
-    port->stop_traffic();
-    result["result"] = "ACK";
+
+    try {
+        port->update_traffic(mul);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    result["result"]["multiplier"] = port->get_multiplier();
 
     return (TREX_RPC_CMD_OK);
 }
 
+/***************************
+ * validate
+ *  
+ * checks that the port
+ * attached streams are 
+ * valid as a program 
+ **************************/
+trex_rpc_cmd_rc_e
+TrexRpcCmdValidate::_run(const Json::Value &params, Json::Value &result) {
+    uint8_t port_id = parse_port(params, result);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    const TrexStreamsGraphObj *graph = NULL;
+
+    try {
+        graph = port->validate();
+    }
+    catch (const TrexException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+    
+
+    result["result"]["rate"]["max_bps"] = graph->get_max_bps();
+    result["result"]["rate"]["max_pps"] = graph->get_max_pps();
+    result["result"]["rate"]["max_line_util"] = graph->get_max_bps() / port->get_port_speed_bps();
+
+    result["result"]["graph"]["expected_duration"] = graph->get_duration();
+    result["result"]["graph"]["events_count"] = (int)graph->get_events().size();
+
+    result["result"]["graph"]["events"] = Json::arrayValue;
+    Json::Value &events_json = result["result"]["graph"]["events"];
+
+    int index = 0;
+    for (const auto &ev : graph->get_events()) {
+        Json::Value ev_json;
+
+        ev_json["time_usec"]  = ev.time;
+        ev_json["diff_bps"]   = ev.diff_bps;
+        ev_json["diff_pps"]   = ev.diff_pps;
+        ev_json["stream_id"]  = ev.stream_id;
+
+        events_json.append(ev_json);
+
+        index++;
+        if (index >= 100) {
+            break;
+        }
+    }
+
+
+    return (TREX_RPC_CMD_OK);
+}
diff --git a/src/rpc-server/commands/trex_rpc_cmds.h b/src/rpc-server/commands/trex_rpc_cmds.h
index 5926a8d8..c22ef390 100644
--- a/src/rpc-server/commands/trex_rpc_cmds.h
+++ b/src/rpc-server/commands/trex_rpc_cmds.h
@@ -77,7 +77,8 @@ TREX_RPC_CMD_DEFINE(TrexRpcCmdRelease,    "release",         1, true);
 /**
  * port commands
  */
-TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortStats, "get_port_stats", 1, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortStats, "get_port_stats", 1, false);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortStatus, "get_port_status", 1, false);
 
 
 /**
@@ -98,12 +99,20 @@ void parse_vm_instr_write_flow_var(const Json::Value &inst, TrexStream *stream,
 );
 
 
-TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStreamList, "get_stream_list", 1, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStreamList, "get_stream_list", 1, false);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetAllStreams, "get_all_streams", 2, false);
 
-TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 2, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 3, false);
 
-TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic, "start_traffic", 1, true);
-TREX_RPC_CMD_DEFINE(TrexRpcCmdStopTraffic, "stop_traffic", 1, true);
 
 
+TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic,  "start_traffic", 3, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdStopTraffic,   "stop_traffic", 1, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdPauseTraffic,  "pause_traffic", 1, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdResumeTraffic, "resume_traffic", 1, true);
+
+TREX_RPC_CMD_DEFINE(TrexRpcCmdUpdateTraffic, "update_traffic", 2, true);
+
+TREX_RPC_CMD_DEFINE(TrexRpcCmdValidate, "validate", 2, false);
+
 #endif /* __TREX_RPC_CMD_H__ */
diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp
index f4d21f2f..46fe499b 100644
--- a/src/rpc-server/trex_rpc_async_server.cpp
+++ b/src/rpc-server/trex_rpc_async_server.cpp
@@ -79,7 +79,7 @@ TrexRpcServerAsync::_rpc_thread_cb() {
         }
 
         /* trigger a full update for stats */
-        TrexStateless::get_instance().update_stats();
+        //get_stateless_obj()->update_stats();
 
         /* done with the lock */
         if (m_lock) {
@@ -87,7 +87,7 @@ TrexRpcServerAsync::_rpc_thread_cb() {
         }
 
         /* encode them to JSON */
-        TrexStateless::get_instance().encode_stats(snapshot);
+        get_stateless_obj()->encode_stats(snapshot);
 
         /* write to string and publish */
         std::string snapshot_str = writer.write(snapshot);
diff --git a/src/rpc-server/trex_rpc_cmd.cpp b/src/rpc-server/trex_rpc_cmd.cpp
index 920a8d30..d4eef1f7 100644
--- a/src/rpc-server/trex_rpc_cmd.cpp
+++ b/src/rpc-server/trex_rpc_cmd.cpp
@@ -61,10 +61,14 @@ TrexRpcCommand::verify_ownership(const Json::Value &params, Json::Value &result)
     std::string handler = parse_string(params, "handler", result);
     uint8_t port_id = parse_port(params, result);
 
-    TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id);
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    if (!port->verify_owner_handler(handler)) {
-        generate_execute_err(result, "invalid handler provided. please pass the handler given when calling 'acquire' or take ownership");
+    if (port->get_owner().is_free()) {
+        generate_execute_err(result, "please acquire the port before modifying port state");
+    }
+
+    if (!port->get_owner().verify(handler)) {
+        generate_execute_err(result, "port is not owned by you or your current executing session");
     }
 }
 
@@ -78,9 +82,9 @@ TrexRpcCommand::parse_port(const Json::Value &params, Json::Value &result) {
 
 void 
 TrexRpcCommand::validate_port_id(uint8_t port_id, Json::Value &result) {
-    if (port_id >= TrexStateless::get_instance().get_port_count()) {
+    if (port_id >= get_stateless_obj()->get_port_count()) {
         std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1;
+        ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1;
         generate_execute_err(result, ss.str());
     }
 }
@@ -92,6 +96,8 @@ TrexRpcCommand::type_to_str(field_type_e type) {
         return "byte";
     case FIELD_TYPE_UINT16:
         return "uint16";
+    case FIELD_TYPE_UINT32:
+        return "uint32";
     case FIELD_TYPE_BOOL:
         return "bool";
     case FIELD_TYPE_INT:
@@ -161,6 +167,18 @@ TrexRpcCommand::parse_uint16(const Json::Value &parent, int index, Json::Value &
     return parent[index].asUInt();
 }
 
+uint32_t 
+TrexRpcCommand::parse_uint32(const Json::Value &parent, const std::string &name, Json::Value &result) {
+    check_field_type(parent, name, FIELD_TYPE_UINT32, result);
+    return parent[name].asUInt();
+}
+
+uint32_t  
+TrexRpcCommand::parse_uint32(const Json::Value &parent, int index, Json::Value &result) {
+    check_field_type(parent, index, FIELD_TYPE_UINT32, result);
+    return parent[index].asUInt();
+}
+
 int
 TrexRpcCommand::parse_int(const Json::Value &parent, const std::string &name, Json::Value &result) {
     check_field_type(parent, name, FIELD_TYPE_INT, result);
@@ -250,6 +268,12 @@ TrexRpcCommand::check_field_type_common(const Json::Value &field, const std::str
         }
         break;
 
+    case FIELD_TYPE_UINT32:
+        if ( (!field.isUInt()) || (field.asUInt() > 0xFFFFFFFF)) {
+            rc = false;
+        }
+        break;
+
     case FIELD_TYPE_BOOL:
         if (!field.isBool()) {
             rc = false;
diff --git a/src/rpc-server/trex_rpc_cmd_api.h b/src/rpc-server/trex_rpc_cmd_api.h
index 3c718eaa..f81981d4 100644
--- a/src/rpc-server/trex_rpc_cmd_api.h
+++ b/src/rpc-server/trex_rpc_cmd_api.h
@@ -99,6 +99,7 @@ protected:
     enum field_type_e {
         FIELD_TYPE_BYTE,
         FIELD_TYPE_UINT16,
+        FIELD_TYPE_UINT32,
         FIELD_TYPE_INT,
         FIELD_TYPE_DOUBLE,
         FIELD_TYPE_BOOL,
@@ -136,6 +137,7 @@ protected:
      */
     uint8_t  parse_byte(const Json::Value &parent, const std::string &name, Json::Value &result);
     uint16_t parse_uint16(const Json::Value &parent, const std::string &name, Json::Value &result);
+    uint32_t parse_uint32(const Json::Value &parent, const std::string &name, Json::Value &result);
     int      parse_int(const Json::Value &parent, const std::string &name, Json::Value &result);
     double   parse_double(const Json::Value &parent, const std::string &name, Json::Value &result);
     bool     parse_bool(const Json::Value &parent, const std::string &name, Json::Value &result);
@@ -145,6 +147,7 @@ protected:
 
     uint8_t  parse_byte(const Json::Value &parent, int index, Json::Value &result);
     uint16_t parse_uint16(const Json::Value &parent, int index, Json::Value &result);
+    uint32_t parse_uint32(const Json::Value &parent, int index, Json::Value &result);
     int      parse_int(const Json::Value &parent, int index, Json::Value &result);
     double   parse_double(const Json::Value &parent, int index, Json::Value &result);
     bool     parse_bool(const Json::Value &parent, int index, Json::Value &result);
@@ -159,7 +162,7 @@ protected:
      * parse a field from choices 
      * 
      */
-    template<typename T> T parse_choice(const Json::Value &params, const std::string &name, std::initializer_list<T> choices, Json::Value &result) {
+    template<typename T> T parse_choice(const Json::Value &params, const std::string &name, const std::initializer_list<T> choices, Json::Value &result) {
         const Json::Value &field = params[name];
 
         if (field == Json::Value::null) {
diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp
index c1c546f3..82c723b7 100644
--- a/src/rpc-server/trex_rpc_cmds_table.cpp
+++ b/src/rpc-server/trex_rpc_cmds_table.cpp
@@ -41,6 +41,8 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() {
     register_command(new TrexRpcCmdAcquire());
     register_command(new TrexRpcCmdRelease());
     register_command(new TrexRpcCmdGetPortStats());
+    register_command(new TrexRpcCmdGetPortStatus());
+    
 
     /* stream commands */
     register_command(new TrexRpcCmdAddStream());
@@ -48,10 +50,18 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() {
     register_command(new TrexRpcCmdRemoveAllStreams());
     register_command(new TrexRpcCmdGetStreamList());
     register_command(new TrexRpcCmdGetStream());
+    register_command(new TrexRpcCmdGetAllStreams());
+
     register_command(new TrexRpcCmdStartTraffic());
     register_command(new TrexRpcCmdStopTraffic());
+    register_command(new TrexRpcCmdPauseTraffic());
+    register_command(new TrexRpcCmdResumeTraffic());
+    register_command(new TrexRpcCmdUpdateTraffic());
+
+    register_command(new TrexRpcCmdValidate());
 }
 
+
 TrexRpcCommandsTable::~TrexRpcCommandsTable() {
     for (auto cmd : m_rpc_cmd_table) {
         delete cmd.second;
diff --git a/src/rpc-server/trex_rpc_req_resp_server.cpp b/src/rpc-server/trex_rpc_req_resp_server.cpp
index 9147f75d..eb7825ac 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.cpp
+++ b/src/rpc-server/trex_rpc_req_resp_server.cpp
@@ -26,6 +26,7 @@ limitations under the License.
 #include <unistd.h>
 #include <sstream>
 #include <iostream>
+#include <assert.h>
 
 #include <zmq.h>
 #include <json/json.h>
@@ -70,28 +71,14 @@ void TrexRpcServerReqRes::_rpc_thread_cb() {
 
     /* server main loop */
     while (m_is_running) {
-        int msg_size = zmq_recv (m_socket, m_msg_buffer, sizeof(m_msg_buffer), 0);
-
-        /* msg_size of -1 is an error - decode it */
-        if (msg_size == -1) {
-            /* normal shutdown and zmq_term was called */
-            if (errno == ETERM) {
-                break;
-            } else {
-                throw TrexRpcException("Unhandled error of zmq_recv");
-            }
-        }
+        std::string request;
 
-        if (msg_size >= sizeof(m_msg_buffer)) {
-            std::stringstream ss;
-            ss << "RPC request of '" << msg_size << "' exceeds maximum message size which is '" << sizeof(m_msg_buffer) << "'";
-            handle_server_error(ss.str());
-            continue;
+        /* get the next request */
+        bool rc = fetch_one_request(request);
+        if (!rc) {
+            break;
         }
 
-        /* transform it to a string */
-        std::string request((const char *)m_msg_buffer, msg_size);
-
         verbose_json("Server Received: ", TrexJsonRpcV2Parser::pretty_json_str(request));
 
         handle_request(request);
@@ -101,6 +88,35 @@ void TrexRpcServerReqRes::_rpc_thread_cb() {
     zmq_close(m_socket);
 }
 
+bool
+TrexRpcServerReqRes::fetch_one_request(std::string &msg) {
+
+    zmq_msg_t zmq_msg;
+    int rc;
+
+    rc = zmq_msg_init(&zmq_msg);
+    assert(rc == 0);
+
+    rc = zmq_msg_recv (&zmq_msg, m_socket, 0);
+
+    if (rc == -1) {
+        zmq_msg_close(&zmq_msg);
+        /* normal shutdown and zmq_term was called */
+        if (errno == ETERM) {
+            return false;
+        } else {
+            throw TrexRpcException("Unhandled error of zmq_recv");
+        }
+    }
+
+    const char *data =  (const char *)zmq_msg_data(&zmq_msg);
+    size_t len = zmq_msg_size(&zmq_msg);
+    msg.append(data, len);
+
+    zmq_msg_close(&zmq_msg);
+    return true;
+}
+
 /**
  * stops the ZMQ based RPC server
  * 
diff --git a/src/rpc-server/trex_rpc_req_resp_server.h b/src/rpc-server/trex_rpc_req_resp_server.h
index 1f638adf..2876206c 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.h
+++ b/src/rpc-server/trex_rpc_req_resp_server.h
@@ -39,14 +39,12 @@ protected:
     void _stop_rpc_thread();
 
 private:
-
+    bool fetch_one_request(std::string &msg);
     void handle_request(const std::string &request);
     void handle_server_error(const std::string &specific_err);
 
-    static const int    RPC_MAX_MSG_SIZE = (20 * 1024);
     void               *m_context;
     void               *m_socket;
-    uint8_t             m_msg_buffer[RPC_MAX_MSG_SIZE];
 };