Improve incoming_data_handler speed

Add test with unexpected bytes in stream

3 files changed, 81 insertions, 17 deletions

diff --git a/src/components/protocol_handler/src/incoming_data_handler.cc b/src/components/protocol_handler/src/incoming_data_handler.cc
index 373a3b2339..9a2c358bbd 100644
--- a/src/components/protocol_handler/src/incoming_data_handler.cc
+++ b/src/components/protocol_handler/src/incoming_data_handler.cc
@@ -122,7 +122,6 @@ RESULT_CODE IncomingDataHandler::CreateFrame(std::vector<uint8_t>& incoming_data
   DCHECK(malformed_occurs);
   *malformed_occurs = false;
   std::vector<uint8_t>::iterator data_it = incoming_data.begin();
-//  uint8_t* data = &incoming_data[0];
   size_t data_size = incoming_data.size();
   while (data_size >= MIN_HEADER_SIZE) {
     header_.deserialize(&*data_it, data_size);
@@ -136,7 +135,7 @@ RESULT_CODE IncomingDataHandler::CreateFrame(std::vector<uint8_t>& incoming_data
       --data_size;
       LOG4CXX_DEBUG(logger_, "Moved to the next byte " << std::hex
                     << static_cast<const void *>(&*data_it));
-      return RESULT_MALFORMED_OCCURS;
+      continue;
     }
     LOG4CXX_DEBUG(logger_, "Payload size " << header_.dataSize);
     const uint32_t packet_size = GetPacketSize(header_);
@@ -150,6 +149,7 @@ RESULT_CODE IncomingDataHandler::CreateFrame(std::vector<uint8_t>& incoming_data
     }
     if (data_size < packet_size) {
       LOG4CXX_DEBUG(logger_, "Packet data is not available yet");
+      incoming_data.erase(incoming_data.begin(), data_it);
       return RESULT_DEFERRED;
     }
     ProtocolFramePtr frame(new protocol_handler::ProtocolPacket(connection_id));
@@ -157,15 +157,15 @@ RESULT_CODE IncomingDataHandler::CreateFrame(std::vector<uint8_t>& incoming_data
         frame->deserializePacket(&*data_it, packet_size);
     if (deserialize_result != RESULT_OK) {
       LOG4CXX_WARN(logger_, "Packet deserialization failed");
+      incoming_data.erase(incoming_data.begin(), data_it);
       return RESULT_FAIL;
     }
     out_frames.push_back(frame);
 
-    incoming_data.erase(data_it, data_it + packet_size);
-//    data = &incoming_data[0];
-    data_it = incoming_data.begin();
-    data_size = incoming_data.size();
+    data_it += packet_size;
+    data_size -= packet_size;
   }
+  incoming_data.erase(incoming_data.begin(), data_it);
   return RESULT_OK;
 }
 }  // namespace protocol_handler
diff --git a/src/components/protocol_handler/test/incoming_data_handler_test.cc b/src/components/protocol_handler/test/incoming_data_handler_test.cc
index 0301193a6a..c47e9dda8b 100644
--- a/src/components/protocol_handler/test/incoming_data_handler_test.cc
+++ b/src/components/protocol_handler/test/incoming_data_handler_test.cc
@@ -194,12 +194,13 @@ TEST_F(IncomingDataHandlerTest, MixedPayloadData_TwoConnections) {
   }
   ProcessData(uid1, &tm_data[0], tm_data.size());
   EXPECT_EQ(RESULT_OK, result_code);
-  EXPECT_EQ(actual_frames.size(), mobile_packets.size());
-  FrameList::const_iterator it2 = mobile_packets.begin();
+  EXPECT_EQ(mobile_packets.size(), actual_frames.size());
+  FrameList::iterator it_exp = mobile_packets.begin();
   for (FrameList::const_iterator it = actual_frames.begin(); it != actual_frames.end();
-       ++it, ++it2) {
+       ++it, ++it_exp) {
     // TODO(EZamakhov): investigate valgrind warning (unitialized value)
-    EXPECT_EQ(**it, **it2);
+    EXPECT_EQ(**it, **it_exp)
+        << "Element number " << std::distance(mobile_packets.begin(), it_exp);
   }
 }
 
@@ -349,6 +350,69 @@ TEST_F(IncomingDataHandlerTest, MalformedPacket_FirstFrame) {
   }
 }
 
+// For Single and First frames Frame info value shall be equal 0x00
+TEST_F(IncomingDataHandlerTest, MalformedPacket_AdditionalByte) {
+  FrameList mobile_packets;
+  // single packet RPC
+  mobile_packets.push_back(
+        new ProtocolPacket(
+          uid1, PROTOCOL_VERSION_1, PROTECTION_OFF, FRAME_TYPE_SINGLE,
+          kRpc, FRAME_DATA_SINGLE, some_session_id, some_data_size,
+          protov1_message_id, some_data));
+  AppendPacketToTMData(*mobile_packets.back());
+  // Add malformed byte
+  tm_data.insert(tm_data.end(), 1, 0x1);
+
+  // consecutive packet Audio
+  mobile_packets.push_back(
+        new ProtocolPacket(
+          uid1, PROTOCOL_VERSION_2, PROTECTION_ON, FRAME_TYPE_CONSECUTIVE,
+          kAudio, FRAME_DATA_LAST_CONSECUTIVE, ++some_session_id, some_data2_size,
+          some_message_id, some_data2));
+  AppendPacketToTMData(*mobile_packets.back());
+  // Add malformed bytes
+  tm_data.insert(tm_data.end(), 2, 0x2);
+
+  // single packet Nav
+  mobile_packets.push_back(
+        new ProtocolPacket(
+          uid1, PROTOCOL_VERSION_3, PROTECTION_ON, FRAME_TYPE_SINGLE,
+          kMobileNav, FRAME_DATA_SINGLE, ++some_session_id, some_data_size,
+          ++some_message_id, some_data));
+  AppendPacketToTMData(*mobile_packets.back());
+  // Add malformed bytes
+  tm_data.insert(tm_data.end(), 3, 0x3);
+
+  // consecutive packet Bulk
+  mobile_packets.push_back(
+        new ProtocolPacket(
+          uid1, PROTOCOL_VERSION_3, PROTECTION_ON, FRAME_TYPE_CONSECUTIVE,
+          kBulk, FRAME_DATA_LAST_CONSECUTIVE, ++some_session_id, some_data2_size,
+          ++some_message_id, some_data2));
+  AppendPacketToTMData(*mobile_packets.back());
+  // Add malformed bytes
+  tm_data.insert(tm_data.end(), 4, 0x4);
+
+  // single packet RPC
+  mobile_packets.push_back(
+        new ProtocolPacket(
+          uid1, PROTOCOL_VERSION_1, PROTECTION_OFF, FRAME_TYPE_CONTROL,
+          kRpc, FRAME_DATA_HEART_BEAT, some_session_id, some_data_size,
+          protov1_message_id, some_data));
+  AppendPacketToTMData(*mobile_packets.back());
+
+  ProcessData(uid1, &tm_data[0], tm_data.size());
+  EXPECT_EQ(RESULT_MALFORMED_OCCURS, result_code);
+  EXPECT_EQ(mobile_packets.size(), actual_frames.size());
+  FrameList::iterator it_exp = mobile_packets.begin();
+  for (FrameList::const_iterator it = actual_frames.begin(); it != actual_frames.end();
+       ++it, ++it_exp) {
+    // TODO(EZamakhov): investigate valgrind warning (unitialized value)
+    EXPECT_EQ(**it, **it_exp)
+        << "Element number " << std::distance(mobile_packets.begin(), it_exp);
+  }
+}
+
 // TODO(EZamakhov): add correctness on handling 2+ connection data
 
 }  // namespace protocol_handler_test
diff --git a/src/components/protocol_handler/test/protocol_handler_tm_test.cc b/src/components/protocol_handler/test/protocol_handler_tm_test.cc
index 74766c2969..89f44d1c4f 100644
--- a/src/components/protocol_handler/test/protocol_handler_tm_test.cc
+++ b/src/components/protocol_handler/test/protocol_handler_tm_test.cc
@@ -60,7 +60,7 @@ using ::testing::Invoke;
 
 class ProtocolHandlerImplTest : public ::testing::Test {
  protected:
-  void IntitProtocolHandlerImpl(
+  void IntiProtocolHandlerImpl(
       const size_t period_msec, const size_t max_messages,
       const size_t malformd_period_msec = 0u,
       const size_t malformd_max_messages = 0u) {
@@ -72,7 +72,7 @@ class ProtocolHandlerImplTest : public ::testing::Test {
     tm_listener = protocol_handler_impl.get();
   }
   void SetUp() OVERRIDE {
-    IntitProtocolHandlerImpl(0u, 0u);
+    IntiProtocolHandlerImpl(0u, 0u);
     connection_id = 0xAu;
     session_id = 0xFFu;
     connection_key = 0xFF00AAu;
@@ -677,7 +677,7 @@ TEST_F(ProtocolHandlerImplTest,
     FloodVerification) {
   const size_t period_msec = 1000;
   const size_t max_messages = 1000;
-  IntitProtocolHandlerImpl(period_msec, max_messages);
+  IntiProtocolHandlerImpl(period_msec, max_messages);
   AddConnection();
   AddSession();
 
@@ -696,7 +696,7 @@ TEST_F(ProtocolHandlerImplTest,
     FloodVerification_ThresholdValue) {
   const size_t period_msec = 1000;
   const size_t max_messages = 1000;
-  IntitProtocolHandlerImpl(period_msec, max_messages);
+  IntiProtocolHandlerImpl(period_msec, max_messages);
   AddConnection();
   AddSession();
 
@@ -711,7 +711,7 @@ TEST_F(ProtocolHandlerImplTest,
     FloodVerification_VideoFrameSkip) {
   const size_t period_msec = 1000;
   const size_t max_messages = 1000;
-  IntitProtocolHandlerImpl(period_msec, max_messages);
+  IntiProtocolHandlerImpl(period_msec, max_messages);
   AddConnection();
   AddSession();
 
@@ -726,7 +726,7 @@ TEST_F(ProtocolHandlerImplTest,
     FloodVerification_AudioFrameSkip) {
   const size_t period_msec = 1000;
   const size_t max_messages = 1000;
-  IntitProtocolHandlerImpl(period_msec, max_messages);
+  IntiProtocolHandlerImpl(period_msec, max_messages);
   AddConnection();
   AddSession();
 
@@ -741,7 +741,7 @@ TEST_F(ProtocolHandlerImplTest,
     FloodVerificationDisable) {
   const size_t period_msec = 0;
   const size_t max_messages = 0;
-  IntitProtocolHandlerImpl(period_msec, max_messages);
+  IntiProtocolHandlerImpl(period_msec, max_messages);
   AddConnection();
   AddSession();