5 files changed, 224 insertions, 62 deletions

diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index b40e996f..1a7132ff 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -29,7 +29,7 @@ limitations under the License.
 #include <iostream>
 #include <unistd.h>
 
-#ifndef TREX_RPC_MOCK_SERVER
+#ifdef RTE_DPDK
     #include <../linux_dpdk/version.h>
 #endif
 
@@ -73,7 +73,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();
@@ -222,12 +222,12 @@ TrexRpcCmdAcquire::_run(const Json::Value &params, Json::Value &result) {
     /* if not free and not you and not force - fail */
     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();
 
     return (TREX_RPC_CMD_OK);
@@ -244,12 +244,12 @@ TrexRpcCmdRelease::_run(const Json::Value &params, Json::Value &result) {
 
     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";
 
     return (TREX_RPC_CMD_OK);
@@ -266,13 +266,13 @@ TrexRpcCmdGetPortStats::_run(const Json::Value &params, Json::Value &result) {
 
     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");
-    }
-
     result["result"]["status"] = port->get_state_as_string();
 
-    port->encode_stats(result["result"]);
+    try {
+        port->encode_stats(result["result"]);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
 
     return (TREX_RPC_CMD_OK);
 }
@@ -303,7 +303,7 @@ TrexRpcCmdSyncUser::_run(const Json::Value &params, Json::Value &result) {
                 owned_port["streams"] = Json::arrayValue;
 
                 std::vector <TrexStream *> streams;
-                port->get_stream_table()->get_object_list(streams);
+                port->get_object_list(streams);
 
                 for (auto stream : streams) {
                     owned_port["streams"].append(stream->get_stream_json());
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index 9854cad7..cdd13ed6 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -115,7 +115,12 @@ TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
     validate_stream(stream, result);
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(stream->m_port_id);
-    port->get_stream_table()->add_stream(stream);
+
+    try {
+        port->add_stream(stream);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
 
     result["result"] = "ACK";
 
@@ -127,7 +132,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 +140,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 +164,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 +215,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);
@@ -293,7 +308,7 @@ TrexRpcCmdAddStream::validate_stream(const TrexStream *stream, Json::Value &resu
     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;
@@ -319,7 +334,7 @@ TrexRpcCmdRemoveStream::_run(const Json::Value &params, Json::Value &result) {
     }
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
-    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;
@@ -327,7 +342,12 @@ 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";
@@ -350,12 +370,18 @@ TrexRpcCmdRemoveAllStreams::_run(const Json::Value &params, Json::Value &result)
         generate_execute_err(result, ss.str());
     }
 
-       TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
-       port->get_stream_table()->remove_and_delete_all_streams();
+    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());
+    }
 
-       result["result"] = "ACK";
 
-       return (TREX_RPC_CMD_OK);
+    result["result"] = "ACK";
+
+    return (TREX_RPC_CMD_OK);
 }
 
 /***************************
@@ -377,7 +403,7 @@ TrexRpcCmdGetStreamList::_run(const Json::Value &params, Json::Value &result) {
 
        TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-       port->get_stream_table()->get_id_list(stream_list);
+       port->get_id_list(stream_list);
 
        Json::Value json_list = Json::arrayValue;
 
@@ -397,8 +423,8 @@ 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);
-
+    uint8_t  port_id   = parse_byte(params, "port_id", result);
+    bool     get_pkt   = parse_bool(params, "get_pkt", result);
     uint32_t stream_id = parse_int(params, "stream_id", result);
 
     if (port_id >= get_stateless_obj()->get_port_count()) {
@@ -409,7 +435,7 @@ TrexRpcCmdGetStream::_run(const Json::Value &params, Json::Value &result) {
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    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 +444,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,8 +462,9 @@ 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);
-    double mul = parse_double(params, "mul", result);
+    uint8_t port_id  = parse_byte(params, "port_id", result);
+    double mul       = parse_double(params, "mul", result);
+    double duration  = parse_double(params, "duration", result);
 
     if (port_id >= get_stateless_obj()->get_port_count()) {
         std::stringstream ss;
@@ -442,37 +474,121 @@ TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    TrexStatelessPort::rc_e rc = port->start_traffic(mul);
+    try {
+        port->start_traffic(mul, duration);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
 
-    if (rc == TrexStatelessPort::RC_OK) {
-        result["result"] = "ACK";
-    } else {
+    result["result"] = "ACK";
+
+    return (TREX_RPC_CMD_OK);
+}
+
+/***************************
+ * 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 >= get_stateless_obj()->get_port_count()) {
         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;
+        ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1;
+        generate_execute_err(result, ss.str());
+    }
+
+    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"] = "ACK";
+
+    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_byte(params, "port_id", result);
+    bool    get_pkt = parse_bool(params, "get_pkt", result);
+     
+    if (port_id >= get_stateless_obj()->get_port_count()) {
+        std::stringstream ss;
+        ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1;
+        generate_execute_err(result, ss.str());
+    }
+
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    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 << stream->m_stream_id;
+
+        streams_json[ss.str()] = j;
+    }
+
+    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_byte(params, "port_id", result);
+
+    if (port_id >= get_stateless_obj()->get_port_count()) {
+        std::stringstream ss;
+        ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1;
         generate_execute_err(result, ss.str());
     }
 
-   return (TREX_RPC_CMD_OK);
+    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"] = "ACK";
+
+    return (TREX_RPC_CMD_OK);
 }
 
 /***************************
- * start traffic on port
+ * resume traffic
  * 
  **************************/
 trex_rpc_cmd_rc_e
-TrexRpcCmdStopTraffic::_run(const Json::Value &params, Json::Value &result) {
-    uint8_t  port_id = parse_byte(params, "port_id", result);
+TrexRpcCmdResumeTraffic::_run(const Json::Value &params, Json::Value &result) {
+
+    uint8_t port_id = parse_byte(params, "port_id", result);
 
     if (port_id >= get_stateless_obj()->get_port_count()) {
         std::stringstream ss;
@@ -482,7 +598,41 @@ TrexRpcCmdStopTraffic::_run(const Json::Value &params, Json::Value &result) {
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
-    port->stop_traffic();
+     try {
+        port->resume_traffic();
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    result["result"] = "ACK";
+
+    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_byte(params, "port_id", result);
+    double mul = parse_double(params, "mul", result);
+
+    if (port_id >= get_stateless_obj()->get_port_count()) {
+        std::stringstream ss;
+        ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1;
+        generate_execute_err(result, ss.str());
+    }
+
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    try {
+        port->update_traffic(mul);
+    } catch (const TrexRpcException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
     result["result"] = "ACK";
 
     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 91c29548..b4f37e3b 100644
--- a/src/rpc-server/commands/trex_rpc_cmds.h
+++ b/src/rpc-server/commands/trex_rpc_cmds.h
@@ -99,11 +99,18 @@ 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(TrexRpcCmdGetAllStreams, "get_all_streams", 2, true);
 
-TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 2, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 3, true);
 
-TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic, "start_traffic", 2, 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(TrexRpcCmdSyncUser, "sync_user", 2, false);
 
diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp
index 46281aff..a65bbccf 100644
--- a/src/rpc-server/trex_rpc_cmds_table.cpp
+++ b/src/rpc-server/trex_rpc_cmds_table.cpp
@@ -50,8 +50,13 @@ 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());
 }
 
 TrexRpcCommandsTable::~TrexRpcCommandsTable() {
diff --git a/src/rpc-server/trex_rpc_req_resp_server.h b/src/rpc-server/trex_rpc_req_resp_server.h
index 1f638adf..bc38c0ef 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.h
+++ b/src/rpc-server/trex_rpc_req_resp_server.h
@@ -43,7 +43,7 @@ private:
     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);
+    static const int    RPC_MAX_MSG_SIZE = (200 * 1024);
     void               *m_context;
     void               *m_socket;
     uint8_t             m_msg_buffer[RPC_MAX_MSG_SIZE];