8 files changed, 176 insertions, 68 deletions

diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index 484cd2b9..6b765aca 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -34,9 +34,6 @@ using namespace std;
 trex_rpc_cmd_rc_e
 TrexRpcCmdGetStatus::_run(const Json::Value &params, Json::Value &result) {
 
-    /* validate count */
-    check_param_count(params, 0, result);
-
     Json::Value &section = result["result"];
 
     #ifndef TREX_RPC_MOCK_SERVER
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index fcd91ab7..25dee501 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -35,65 +35,148 @@ using namespace std;
 trex_rpc_cmd_rc_e
 TrexRpcCmdAddStream::_run(const Json::Value &params, Json::Value &result) {
 
-    TrexStream *stream;
-
-    check_param_count(params, 1, result);
-
     const Json::Value &section = parse_object(params, "stream", result);
 
     /* get the type of the stream */
     const Json::Value &mode = parse_object(section, "mode", result);
     string type = parse_string(mode, "type", result);
 
+    TrexStream *stream = allocate_new_stream(section, result);
+
+    /* create a new steram and populate it */
+   
+    stream->m_isg_usec  = parse_double(section, "Is", result);
+
+    stream->m_next_stream_id = parse_int(section, "next_stream_id", result);
+    stream->m_loop_count     = parse_int(section, "loop_count", result);
+
+    const Json::Value &pkt = parse_array(section, "packet", result);
+
+    /* fetch the packet from the message */
+
+    stream->m_pkt_len = pkt.size();
+    stream->m_pkt = new uint8_t[pkt.size()];
+    if (!stream->m_pkt) {
+        generate_internal_err(result, "unable to allocate memory");
+    }
+
+    for (int i = 0; i < pkt.size(); i++) {
+        stream->m_pkt[i] = parse_byte(pkt, i, result);
+    }
+
+    /* make sure this is a valid stream to add */
+    validate_stream(stream, result);
+
+    TrexStatelessPort *port = get_trex_stateless()->get_port_by_id(stream->m_port_id);
+    port->get_stream_table()->add_stream(stream);
+
+    result["result"] = "ACK";
+
+    return (TREX_RPC_CMD_OK);
+}
+
+
+
+TrexStream *
+TrexRpcCmdAddStream::allocate_new_stream(const Json::Value &section, Json::Value &result) {
+
+    uint8_t  port_id    = parse_int(section, "port_id", result);
+    uint32_t stream_id  = parse_int(section, "stream_id", result);
+
+    TrexStream *stream;
+
+    const Json::Value &mode = parse_object(section, "mode", result);
+    std::string type = parse_string(mode, "type", result);
+
     if (type == "continuous") {
-        stream = new TrexStreamContinuous();
+
+        uint32_t pps = parse_int(mode, "pps", result);
+        stream = new TrexStreamContinuous(port_id, stream_id, pps);
+
     } else if (type == "single_burst") {
-        stream = new TrexStreamSingleBurst();
+
+        uint32_t pps = parse_int(mode, "pps", result);
+        uint32_t packets = parse_int(type, "packets", result);
+
+        stream = new TrexStreamSingleBurst(port_id, stream_id, pps, packets);
+
     } else if (type == "multi_burst") {
-        stream = new TrexStreamMultiBurst();
+
+        uint32_t  pps           = parse_int(mode, "pps", result);
+        double    ibg_usec      = parse_double(mode, "ibg", result);
+        uint32_t  num_bursts    = parse_int(mode, "number_of_bursts", result);
+        uint32_t  pkt_per_burst = parse_int(mode, "pkt_per_burst", result);
+
+        stream = new TrexStreamMultiBurst(port_id, stream_id, pps, ibg_usec, num_bursts, pkt_per_burst);
+        
+
     } else {
-        generate_err(result, "bad stream type provided: '" + type + "'");
+        generate_parse_err(result, "bad stream type provided: '" + type + "'");
     }
 
     if (!stream) {
         generate_internal_err(result, "unable to allocate memory");
     }
 
-    /* create a new steram and populate it */
-    stream->stream_id = parse_int(section, "stream_id", result);
-    stream->port_id   = parse_int(section, "port_id", result);
-    stream->isg_usec  = parse_double(section, "Is", result);
+    return (stream);
 
-    stream->next_stream_id = parse_int(section, "next_stream_id", result);
-    stream->loop_count     = parse_int(section, "loop_count", result);
+}
 
-    const Json::Value &pkt = parse_array(section, "packet", result);
+void
+TrexRpcCmdAddStream::validate_stream(const TrexStream *stream, Json::Value &result) {
 
-    if ( (pkt.size() < TrexStream::MIN_PKT_SIZE_BYTES) || (pkt.size() > TrexStream::MAX_PKT_SIZE_BYTES) ) {
-        generate_err(result, "bad packet size provided: should be between 64B and 9K"); 
+    /* check packet size */
+    if ( (stream->m_pkt_len < TrexStream::MIN_PKT_SIZE_BYTES) || (stream->m_pkt_len > TrexStream::MAX_PKT_SIZE_BYTES) ) {
+        std::stringstream ss;
+        ss << "bad packet size provided: should be between " << TrexStream::MIN_PKT_SIZE_BYTES << " and " << TrexStream::MAX_PKT_SIZE_BYTES;
+        delete stream;
+        generate_execute_err(result, ss.str()); 
     }
 
-    stream->pkt = new uint8_t[pkt.size()];
-    if (!stream->pkt) {
-        generate_internal_err(result, "unable to allocate memory");
+    /* port id should be between 0 and count - 1 */
+    if (stream->m_port_id >= get_trex_stateless()->get_port_count()) {
+        std::stringstream ss;
+        ss << "invalid port id - should be between 0 and " << (int)get_trex_stateless()->get_port_count() - 1;
+        delete stream;
+        generate_execute_err(result, ss.str());
     }
 
-    for (int i = 0; i < pkt.size(); i++) {
-        stream->pkt[i] = parse_byte(pkt, i, result);
+     /* add the stream to the port's stream table */
+    TrexStatelessPort * port = get_trex_stateless()->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)) {
+        std::stringstream ss;
+        ss << "stream " << stream->m_stream_id << " already exists";
+        delete stream;
+        generate_execute_err(result, ss.str());
     }
 
-    /* register the stream to the port */
+}
+
+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);
+
 
-    /* port id should be between 0 and count - 1 */
-    if (stream->port_id >= get_trex_stateless()->get_port_count()) {
+    if (port_id >= get_trex_stateless()->get_port_count()) {
         std::stringstream ss;
-        ss << "invalid port id - should be between 0 and " << get_trex_stateless()->get_port_count();
-        generate_err(result, ss.str());
+        ss << "invalid port id - should be between 0 and " << (int)get_trex_stateless()->get_port_count() - 1;
+        generate_execute_err(result, ss.str());
     }
 
-    TrexStatelessPort * port = get_trex_stateless()->get_port_by_id(stream->port_id);
-    port->get_stream_table()->add_stream(stream);
+    TrexStatelessPort *port = get_trex_stateless()->get_port_by_id(port_id);
+    TrexStream *stream  = port->get_stream_table()->get_stream_by_id(stream_id);
 
-    return (TREX_RPC_CMD_OK);
+    if (!stream) {
+        std::stringstream ss;
+        ss << "stream " << stream_id << " does not exists";
+        generate_execute_err(result, ss.str());
+    }
+
+    port->get_stream_table()->remove_stream(stream);
+
+    result["result"] = "ACK";
 }
 
diff --git a/src/rpc-server/commands/trex_rpc_cmd_test.cpp b/src/rpc-server/commands/trex_rpc_cmd_test.cpp
index 473cbb70..382279ba 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_test.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_test.cpp
@@ -35,8 +35,6 @@ TrexRpcCmdTestAdd::_run(const Json::Value &params, Json::Value &result) {
     const Json::Value &x = params["x"];
     const Json::Value &y = params["y"];
     
-    check_param_count(params, 2, result);
-
     result["result"] = parse_int(params, "x", result) + parse_int(params, "y", result);
 
     return (TREX_RPC_CMD_OK);
@@ -53,8 +51,6 @@ TrexRpcCmdTestSub::_run(const Json::Value &params, Json::Value &result) {
     const Json::Value &x = params["x"];
     const Json::Value &y = params["y"];
         
-    check_param_count(params, 2, result);
-
     result["result"] = parse_int(params, "x", result) - parse_int(params, "y", result);
 
     return (TREX_RPC_CMD_OK);
@@ -66,9 +62,6 @@ TrexRpcCmdTestSub::_run(const Json::Value &params, Json::Value &result) {
 trex_rpc_cmd_rc_e 
 TrexRpcCmdPing::_run(const Json::Value &params, Json::Value &result) {
 
-    /* validate count */
-    check_param_count(params, 0, result);
-
     result["result"] = "ACK";
     return (TREX_RPC_CMD_OK);
 }
@@ -80,9 +73,6 @@ trex_rpc_cmd_rc_e
 TrexRpcCmdGetReg::_run(const Json::Value &params, Json::Value &result) {
     vector<string> cmds;
 
-    /* validate count */
-    check_param_count(params, 0, result);
-
     TrexRpcCommandsTable::get_instance().query(cmds);
 
     Json::Value test = Json::arrayValue;
diff --git a/src/rpc-server/commands/trex_rpc_cmds.h b/src/rpc-server/commands/trex_rpc_cmds.h
index 5c425856..7ec8aba3 100644
--- a/src/rpc-server/commands/trex_rpc_cmds.h
+++ b/src/rpc-server/commands/trex_rpc_cmds.h
@@ -25,6 +25,8 @@ limitations under the License.
 #include <trex_rpc_cmd_api.h>
 #include <json/json.h>
 
+class TrexStream;
+
 /* all the RPC commands decl. goes here */
 
 /******************* test section ************/
@@ -32,31 +34,43 @@ limitations under the License.
 /**
  * syntactic sugar for creating a simple command
  */
-#define TREX_RPC_CMD_DEFINE(class_name, cmd_name)                                       \
-    class class_name : public TrexRpcCommand {                                          \
-    public:                                                                             \
-        class_name () : TrexRpcCommand(cmd_name) {}                                     \
-    protected:                                                                          \
-        virtual trex_rpc_cmd_rc_e _run(const Json::Value &params, Json::Value &result); \
+
+#define TREX_RPC_CMD_DEFINE_EXTENED(class_name, cmd_name, param_count, ext)                               \
+    class class_name : public TrexRpcCommand {                                                            \
+    public:                                                                                               \
+        class_name () : TrexRpcCommand(cmd_name, param_count) {}                                          \
+    protected:                                                                                            \
+        virtual trex_rpc_cmd_rc_e _run(const Json::Value &params, Json::Value &result);                   \
+        ext                                                                                               \
     }
 
+#define TREX_RPC_CMD_DEFINE(class_name, cmd_name, param_count) TREX_RPC_CMD_DEFINE_EXTENED(class_name, cmd_name, param_count, ;)
 
 /**
  * test cmds
  */
-TREX_RPC_CMD_DEFINE(TrexRpcCmdTestAdd,    "test_add");
-TREX_RPC_CMD_DEFINE(TrexRpcCmdTestSub,    "test_sub");
+TREX_RPC_CMD_DEFINE(TrexRpcCmdTestAdd,    "test_add", 2);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdTestSub,    "test_sub", 2);
 
 /**
  * general cmds
  */
-TREX_RPC_CMD_DEFINE(TrexRpcCmdPing,       "ping");
-TREX_RPC_CMD_DEFINE(TrexRpcCmdGetReg,     "get_reg_cmds");
-TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStatus,  "get_status");
+TREX_RPC_CMD_DEFINE(TrexRpcCmdPing,       "ping",            0);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetReg,     "get_reg_cmds",    0);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStatus,  "get_status",      0);
 
 /**
  * stream cmds
  */
-TREX_RPC_CMD_DEFINE(TrexRpcCmdAddStream,  "add_stream");
+TREX_RPC_CMD_DEFINE(TrexRpcCmdRemoveStream,   "remove_stream",   2);
+
+TREX_RPC_CMD_DEFINE_EXTENED(TrexRpcCmdAddStream, "add_stream", 1,
+
+/* extended part */
+TrexStream * allocate_new_stream(const Json::Value &section, Json::Value &result);
+void validate_stream(const TrexStream *stream, Json::Value &result);
+
+);
+
 
 #endif /* __TREX_RPC_CMD_H__ */
diff --git a/src/rpc-server/trex_rpc_cmd.cpp b/src/rpc-server/trex_rpc_cmd.cpp
index 1ba36e32..6988cba7 100644
--- a/src/rpc-server/trex_rpc_cmd.cpp
+++ b/src/rpc-server/trex_rpc_cmd.cpp
@@ -26,6 +26,7 @@ TrexRpcCommand::run(const Json::Value &params, Json::Value &result) {
 
     /* the internal run can throw a parser error / other error */
     try {
+        check_param_count(params, m_param_count, result);
         rc = _run(params, result);
     } catch (TrexRpcCommandException &e) {
         return e.get_rc();
@@ -40,7 +41,7 @@ TrexRpcCommand::check_param_count(const Json::Value &params, int expected, Json:
     if (params.size() != expected) {
         std::stringstream ss;
         ss << "method expects '" << expected << "' paramteres, '" << params.size() << "' provided";
-        generate_err(result, ss.str());
+        generate_parse_err(result, ss.str());
     }
 }
 
@@ -177,7 +178,7 @@ TrexRpcCommand::check_field_type_common(const Json::Value &field, const std::str
     /* first check if field exists */
     if (field == Json::Value::null) {
         ss << "field '" << name << "' is missing";
-        generate_err(result, ss.str());
+        generate_parse_err(result, ss.str());
     }
 
     bool rc = true;
@@ -232,15 +233,15 @@ TrexRpcCommand::check_field_type_common(const Json::Value &field, const std::str
     }
     if (!rc) {
         ss << "error at offset: " << field.getOffsetStart() << " - '" << name << "' is '" << json_type_to_name(field) << "', expecting '" << type_to_str(type) << "'";
-        generate_err(result, ss.str());
+        generate_parse_err(result, ss.str());
     }
 
 }
 
 void 
-TrexRpcCommand::generate_err(Json::Value &result, const std::string &msg) {
+TrexRpcCommand::generate_parse_err(Json::Value &result, const std::string &msg) {
     result["specific_err"] = msg;
-    throw (TrexRpcCommandException(TREX_RPC_CMD_PARAM_PARSE_ERR));
+    throw (TrexRpcCommandException(TREX_RPC_CMD_PARSE_ERR));
 }
 
 void 
@@ -249,3 +250,9 @@ TrexRpcCommand::generate_internal_err(Json::Value &result, const std::string &ms
     throw (TrexRpcCommandException(TREX_RPC_CMD_INTERNAL_ERR));
 }
 
+void 
+TrexRpcCommand::generate_execute_err(Json::Value &result, const std::string &msg) {
+    result["specific_err"] = msg;
+    throw (TrexRpcCommandException(TREX_RPC_CMD_EXECUTE_ERR));
+}
+
diff --git a/src/rpc-server/trex_rpc_cmd_api.h b/src/rpc-server/trex_rpc_cmd_api.h
index 40f839df..da895809 100644
--- a/src/rpc-server/trex_rpc_cmd_api.h
+++ b/src/rpc-server/trex_rpc_cmd_api.h
@@ -32,8 +32,8 @@ limitations under the License.
  */
 typedef enum trex_rpc_cmd_rc_ {
     TREX_RPC_CMD_OK,
-    TREX_RPC_CMD_PARAM_COUNT_ERR = 1,
-    TREX_RPC_CMD_PARAM_PARSE_ERR,
+    TREX_RPC_CMD_PARSE_ERR,
+    TREX_RPC_CMD_EXECUTE_ERR,
     TREX_RPC_CMD_INTERNAL_ERR
 } trex_rpc_cmd_rc_e;
 
@@ -68,7 +68,7 @@ public:
     /**
      * method name and params
      */
-    TrexRpcCommand(const std::string &method_name) : m_name(method_name) {
+    TrexRpcCommand(const std::string &method_name, int param_count) : m_name(method_name), m_param_count(param_count) {
 
     }
 
@@ -142,15 +142,22 @@ protected:
      * error generating functions
      * 
      */
-    void generate_err(Json::Value &result, const std::string &msg);
+    void generate_parse_err(Json::Value &result, const std::string &msg);
 
 
     /**
+     * method execute error
+     * 
+     */
+    void generate_execute_err(Json::Value &result, const std::string &msg);
+
+    /**
      * internal error
      * 
      */
     void generate_internal_err(Json::Value &result, const std::string &msg);
 
+
     /**
      * translate enum to string
      * 
@@ -164,7 +171,8 @@ protected:
     const char * json_type_to_name(const Json::Value &value);
 
     /* RPC command name */
-    std::string m_name;
+    std::string   m_name;
+    int           m_param_count;
 };
 
 #endif /* __TREX_RPC_CMD_API_H__ */
diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp
index ac419bfd..e586c3d6 100644
--- a/src/rpc-server/trex_rpc_cmds_table.cpp
+++ b/src/rpc-server/trex_rpc_cmds_table.cpp
@@ -36,6 +36,7 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() {
 
     /* stream commands */
     register_command(new TrexRpcCmdAddStream());
+    register_command(new TrexRpcCmdRemoveStream());
 }
 
 TrexRpcCommandsTable::~TrexRpcCommandsTable() {
diff --git a/src/rpc-server/trex_rpc_jsonrpc_v2_parser.cpp b/src/rpc-server/trex_rpc_jsonrpc_v2_parser.cpp
index bf6056d5..3831bb37 100644
--- a/src/rpc-server/trex_rpc_jsonrpc_v2_parser.cpp
+++ b/src/rpc-server/trex_rpc_jsonrpc_v2_parser.cpp
@@ -37,7 +37,10 @@ enum {
     JSONRPC_V2_ERR_INVALID_REQ        = -32600,
     JSONRPC_V2_ERR_METHOD_NOT_FOUND   = -32601,
     JSONRPC_V2_ERR_INVALID_PARAMS     = -32602,
-    JSONRPC_V2_ERR_INTERNAL_ERROR     = -32603
+    JSONRPC_V2_ERR_INTERNAL_ERROR     = -32603,
+
+    /* specific server errors */
+    JSONRPC_V2_ERR_EXECUTE_ERROR      = -32000,
 };
 
 
@@ -78,13 +81,18 @@ public:
             response["result"] = result["result"];
             break;
 
-        case TREX_RPC_CMD_PARAM_COUNT_ERR:
-        case TREX_RPC_CMD_PARAM_PARSE_ERR:
+        case TREX_RPC_CMD_PARSE_ERR:
             response["error"]["code"]          = JSONRPC_V2_ERR_INVALID_PARAMS;
             response["error"]["message"]       = "Bad paramters for method";
             response["error"]["specific_err"]  = result["specific_err"];
             break;
 
+        case TREX_RPC_CMD_EXECUTE_ERR:
+            response["error"]["code"]          = JSONRPC_V2_ERR_EXECUTE_ERROR;
+            response["error"]["message"]       = "Failed To Execute Method";
+            response["error"]["specific_err"]  = result["specific_err"];
+            break;
+
         case TREX_RPC_CMD_INTERNAL_ERR:
             response["error"]["code"]          = JSONRPC_V2_ERR_INTERNAL_ERROR;
             response["error"]["message"]       = "Internal Server Error";