added "-def" option to read and write commands for using inline message definition (fixes #154), added support for replacing messages during load to FileReader and subclasses, added MessageMap::remove(), prepared decode/encode commands

src/lib/ebus/data.cpp

@@ -1130,7 +1130,7 @@ result_t DataFieldSet::write(char separator, size_t offset, istringstream* input
 
 
 DataFieldTemplates::DataFieldTemplates(const DataFieldTemplates& other)
-    : MappedFileReader::MappedFileReader(false) {
+: MappedFileReader::MappedFileReader(false) {
   for (const auto it : other.m_fieldsByName) {
     m_fieldsByName[it.first] = it.second->clone();
   }
@@ -1250,7 +1250,7 @@ result_t DataFieldTemplates::getFieldMap(const string& preferLanguage, vector<st
 }
 
 result_t DataFieldTemplates::addFromFile(const string& filename, unsigned int lineNo, map<string, string>* row,
-    vector< map<string, string> >* subRows, string* errorDescription) {
+    vector< map<string, string> >* subRows, string* errorDescription, bool replace) {
   string name = (*row)["name"];  // required
   string firstFieldName;
   size_t colon = name.find(':');
@@ -1271,7 +1271,7 @@ result_t DataFieldTemplates::addFromFile(const string& filename, unsigned int li
   if (result != RESULT_OK) {
     return result;
   }
-  result = add(field, name, true);
+  result = add(field, name, replace);
   if (result == RESULT_ERR_DUPLICATE_NAME) {
     *errorDescription = name;
   }

src/lib/ebus/data.h

@@ -746,7 +746,7 @@ class DataFieldTemplates : public MappedFileReader {
 
   // @copydoc
   result_t addFromFile(const string& filename, unsigned int lineNo, map<string, string>* row,
-      vector< map<string, string> >* subRows, string* errorDescription) override;
+      vector< map<string, string> >* subRows, string* errorDescription, bool replace) override;
 
   /**
    * Gets the template @a DataField instance with the specified name.

src/lib/ebus/filereader.cpp

@@ -60,7 +60,7 @@ istream* FileReader::openFile(const string& filename, string* errorDescription,
 }
 
 result_t FileReader::readFromStream(istream* stream, const string& filename, const time_t& mtime, bool verbose,
-    map<string, string>* defaults, string* errorDescription, size_t* hash, size_t* size) {
+    map<string, string>* defaults, string* errorDescription, bool replace, size_t* hash, size_t* size) {
   if (hash) {
     *hash = 0;
   }
@@ -71,20 +71,20 @@ result_t FileReader::readFromStream(istream* stream, const string& filename, con
   vector<string> row;
   result_t result = RESULT_OK;
   while (stream->peek() != EOF && result == RESULT_OK) {
-    result = readLineFromStream(stream, filename, verbose, &lineNo, &row, errorDescription, hash, size);
+    result = readLineFromStream(stream, filename, verbose, &lineNo, &row, errorDescription, replace, hash, size);
   }
   return result;
 }
 
 result_t FileReader::readLineFromStream(istream* stream, const string& filename, bool verbose,
-    unsigned int* lineNo, vector<string>* row, string* errorDescription, size_t* hash, size_t* size) {
+    unsigned int* lineNo, vector<string>* row, string* errorDescription, bool replace, size_t* hash, size_t* size) {
   result_t result;
   if (!splitFields(stream, row, lineNo, hash, size)) {
     *errorDescription = "blank line";
     result = RESULT_ERR_EOF;
   } else {
     *errorDescription = "";
-    result = addFromFile(filename, *lineNo, row, errorDescription);
+    result = addFromFile(filename, *lineNo, row, errorDescription, replace);
   }
   if (result != RESULT_OK) {
     if (!errorDescription->empty()) {
@@ -242,7 +242,7 @@ const string MappedFileReader::normalizeLanguage(const string& lang) {
 }
 
 result_t MappedFileReader::readFromStream(istream* stream, const string& filename, const time_t& mtime, bool verbose,
-    map<string, string>* defaults, string* errorDescription, size_t* hash, size_t* size) {
+    map<string, string>* defaults, string* errorDescription, bool replace, size_t* hash, size_t* size) {
   m_mutex.lock();
   m_columnNames.clear();
   m_lastDefaults.clear();
@@ -253,13 +253,14 @@ result_t MappedFileReader::readFromStream(istream* stream, const string& filenam
   size_t lastSep = filename.find_last_of('/');
   string defaultsPart = lastSep == string::npos ? filename : filename.substr(lastSep+1);
   extractDefaultsFromFilename(defaultsPart, &m_lastDefaults[""]);
-  result_t result = FileReader::readFromStream(stream, filename, mtime, verbose, defaults, errorDescription, hash, size);
+  result_t result
+  = FileReader::readFromStream(stream, filename, mtime, verbose, defaults, errorDescription, replace, hash, size);
   m_mutex.unlock();
   return result;
 }
 
 result_t MappedFileReader::addFromFile(const string& filename, unsigned int lineNo, vector<string>* row,
-    string* errorDescription) {
+    string* errorDescription, bool replace) {
   result_t result;
   if (lineNo == 1) {  // first line defines column names
     result = getFieldMap(m_preferLanguage, row, errorDescription);
@@ -325,7 +326,7 @@ result_t MappedFileReader::addFromFile(const string& filename, unsigned int line
   if (isDefault) {
     return addDefaultFromFile(filename, lineNo, &rowMapped, &subRowsMapped, errorDescription);
   }
-  return addFromFile(filename, lineNo, &rowMapped, &subRowsMapped, errorDescription);
+  return addFromFile(filename, lineNo, &rowMapped, &subRowsMapped, errorDescription, replace);
 }
 
 const string MappedFileReader::combineRow(const map<string, string>& row) {

src/lib/ebus/filereader.h

@@ -92,12 +92,14 @@ class FileReader {
    * @param verbose whether to verbosely log problems.
    * @param defaults the default values by name (potentially overwritten by file name), or NULL to not use defaults.
    * @param errorDescription a string in which to store the error description in case of error.
+   * @param replace whether to replace an already existing entry.
    * @param hash optional pointer to a @a size_t value for storing the hash of the file, or NULL.
    * @param size optional pointer to a @a size_t value for storing the normalized size of the file, or NULL.
    * @return @a RESULT_OK on success, or an error code.
    */
   virtual result_t readFromStream(istream* stream, const string& filename, const time_t& mtime, bool verbose,
-      map<string, string>* defaults, string* errorDescription, size_t* hash = NULL, size_t* size = NULL);
+      map<string, string>* defaults, string* errorDescription, bool replace = false, size_t* hash = NULL,
+      size_t* size = NULL);
 
   /**
    * Read a single line definition from the stream.
@@ -107,12 +109,13 @@ class FileReader {
    * @param lineNo the last line number (incremented with each line read).
    * @param row the definition row to clear and update with the read data (for performance reasons only).
    * @param errorDescription a string in which to store the error description in case of error.
+   * @param replace whether to replace an already existing entry.
    * @param hash optional pointer to a @a size_t value for updating with the hash of the line, or NULL.
    * @param size optional pointer to a @a size_t value for updating with the normalized length of the line, or NULL.
    * @return @a RESULT_OK on success, or an error code.
    */
   virtual result_t readLineFromStream(istream* stream, const string& filename, bool verbose,
-      unsigned int* lineNo, vector<string>* row, string* errorDescription, size_t* hash, size_t* size);
+      unsigned int* lineNo, vector<string>* row, string* errorDescription, bool replace, size_t* hash, size_t* size);
 
   /**
    * Add a definition that was read from a file.
@@ -120,10 +123,11 @@ class FileReader {
    * @param lineNo the current line number in the file being read.
    * @param row the definition row (allowed to be modified).
    * @param errorDescription a string in which to store the error description in case of error.
+   * @param replace whether to replace an already existing entry.
    * @return @a RESULT_OK on success, or an error code.
    */
   virtual result_t addFromFile(const string& filename, unsigned int lineNo, vector<string>* row,
-      string* errorDescription) = 0;
+      string* errorDescription, bool replace) = 0;
 
   /**
    * Left and right trim the string.
@@ -205,7 +209,8 @@ class MappedFileReader : public FileReader {
 
   // @copydoc
   result_t readFromStream(istream* stream, const string& filename, const time_t& mtime, bool verbose,
-      map<string, string>* defaults, string* errorDescription, size_t* hash = NULL, size_t* size = NULL) override;
+      map<string, string>* defaults, string* errorDescription, bool replace = false, size_t* hash = NULL,
+      size_t* size = NULL) override;
 
   /**
    * Extract default values from the file name.
@@ -223,7 +228,7 @@ class MappedFileReader : public FileReader {
 
   // @copydoc
   result_t addFromFile(const string& filename, unsigned int lineNo, vector<string>* row,
-      string* errorDescription) override;
+      string* errorDescription, bool replace) override;
 
   /**
    * Get the field mapping from the given first line.
@@ -258,10 +263,11 @@ class MappedFileReader : public FileReader {
    * @param row the main definition row by field name (may be modified).
    * @param subRows the sub definition rows, each by field name (may be modified).
    * @param errorDescription a string in which to store the error description in case of error.
+   * @param replace whether to replace an already existing entry.
    * @return @a RESULT_OK on success, or an error code.
    */
   virtual result_t addFromFile(const string& filename, unsigned int lineNo, map<string, string>* row,
-      vector< map<string, string> >* subRows, string* errorDescription) = 0;
+      vector< map<string, string> >* subRows, string* errorDescription, bool replace = false) = 0;
 
   /**
    * @return a reference to all previously extracted default values by type and field name.

src/lib/ebus/message.cpp

@@ -82,7 +82,7 @@ static const char* defaultMessageFieldMap[] = {  // access level not included in
 extern DataFieldTemplates* getTemplates(const string& filename);
 
 extern result_t loadDefinitionsFromConfigPath(FileReader* reader, const string& filename, bool verbose,
-    map<string, string>* defaults, string* errorDescription);
+    map<string, string>* defaults, string* errorDescription, bool replace = false);
 
 
 Message::Message(const string& circuit, const string& level, const string& name,
@@ -1733,22 +1733,35 @@ result_t LoadInstruction::execute(MessageMap* messages, ostringstream* log) {
 
 vector<string> MessageMap::s_noFiles;
 
-result_t MessageMap::add(bool storeByName, Message* message) {
+result_t MessageMap::add(bool storeByName, Message* message, bool replace) {
   uint64_t key = message->getKey();
   bool conditional = message->isConditional();
   if (!m_addAll) {
+    lock();
     const auto keyIt = m_messagesByKey.find(key);
     if (keyIt != m_messagesByKey.end()) {
-      Message* other = getFirstAvailable(keyIt->second, message);
-      if (other != NULL) {
-        if (!conditional) {
-          return RESULT_ERR_DUPLICATE;  // duplicate key
+      if (replace) {
+        vector<Message*> removeMessages;
+        for (auto other : keyIt->second) {
+          if (!other || !message->checkId(*other)) {
+            continue;
+          }
+          if (!conditional || !other->isConditional() || other->m_condition == message->m_condition) {
+            removeMessages.push_back(other);
+          }
         }
-        if (!other->isConditional()) {
+        for (auto other : removeMessages) {
+          remove(other);
+        }
+      } else {
+        Message *other = getFirstAvailable(keyIt->second, message);
+        if (other != NULL && (!conditional || !other->isConditional())) {
+          unlock();
           return RESULT_ERR_DUPLICATE;  // duplicate key
         }
       }
     }
+    unlock();
   }
   bool isPassive = message->isPassive();
   if (storeByName) {
@@ -1763,13 +1776,29 @@ result_t MessageMap::add(bool storeByName, Message* message) {
     string suffix = FIELD_SEPARATOR + name + (isPassive ? "P" : (isWrite ? "W" : "R"));
     string nameKey = circuit + suffix;
     if (!m_addAll) {
+      lock();
       const auto nameIt = m_messagesByName.find(nameKey);
       if (nameIt != m_messagesByName.end()) {
         vector<Message*>* messages = &nameIt->second;
-        if (!message->isConditional() || !messages->front()->isConditional()) {
+        if (replace) {
+          vector<Message*> removeMessages;
+          for (auto other : *messages) {
+            if (!other) {
+              continue;
+            }
+            if (!conditional || !other->isConditional() || other->m_condition == message->m_condition) {
+              removeMessages.push_back(other);
+            }
+          }
+          for (auto other : removeMessages) {
+            remove(other);
+          }
+        } else if (!conditional || !messages->front()->isConditional()) {
+          unlock();
           return RESULT_ERR_DUPLICATE_NAME;  // duplicate key
         }
       }
+      unlock();
     }
     m_messagesByName[nameKey].push_back(message);
     nameKey = suffix;  // also store without circuit
@@ -1808,6 +1837,71 @@ result_t MessageMap::add(bool storeByName, Message* message) {
   return RESULT_OK;
 }
 
+void MessageMap::remove(Message* message) {
+  if (message == NULL) {
+    return;
+  }
+  lock();
+  uint64_t key = message->getKey();
+  bool conditional = message->isConditional();
+  const auto keyIt = m_messagesByKey.find(key);
+  bool deleted = false;
+  if (keyIt != m_messagesByKey.end()) {
+    vector<Message*> messages = keyIt->second;
+    for (auto it = messages.begin(); it != messages.end(); ) {
+      Message* other = *it;
+      if (other == message) {
+        if (!deleted) {
+          deleted = true;
+          delete(other);
+        }
+        messages.erase(it);
+      } else {
+        ++it;
+      }
+    }
+    if (messages.empty()) {
+      m_messagesByKey.erase(keyIt);
+    }
+  }
+  bool storedByName = false;
+  for (auto nameIt = m_messagesByName.begin(); nameIt != m_messagesByName.end(); ) {
+    vector<Message*> messages = nameIt->second;
+    for (auto it = messages.begin(); it != messages.end(); ) {
+      Message* other = *it;
+      if (other == message) {
+        storedByName = true;
+        if (!deleted) {
+          deleted = true;
+          delete(other);
+        }
+        messages.erase(it);
+      } else {
+        ++it;
+      }
+    }
+    if (messages.empty()) {
+      m_messagesByName.erase(nameIt);
+    } else {
+      ++nameIt;
+    }
+  }
+  if (storedByName) {
+    bool isPassive = message->isPassive();
+    m_messageCount--;
+    if (conditional) {
+      m_conditionalMessageCount--;
+    }
+    if (isPassive) {
+      m_passiveMessageCount--;
+    }
+  }
+  if (message->getPollPriority() > 0) {
+    m_pollMessages.remove(message);
+  }
+  unlock();
+}
+
 result_t MessageMap::getFieldMap(const string& preferLanguage, vector<string>* row, string* errorDescription) const {
   // type,circuit,name,[comment],[QQ],ZZ,PBSB,[ID],*name,[part],type,divisor/values,unit,comment
   // minimum: type,name,PBSB,*type
@@ -2105,7 +2199,7 @@ bool MessageMap::extractDefaultsFromFilename(const string& filename, map<string,
 }
 
 result_t MessageMap::readFromStream(istream* stream, const string& filename, const time_t& mtime, bool verbose,
-    map<string, string>* defaults, string* errorDescription, size_t* hash, size_t* size) {
+    map<string, string>* defaults, string* errorDescription, bool replace, size_t* hash, size_t* size) {
   size_t localHash, localSize;
   if (!hash) {
     hash = &localHash;
@@ -2113,8 +2207,8 @@ result_t MessageMap::readFromStream(istream* stream, const string& filename, con
   if (!size) {
     size = &localSize;
   }
-  result_t result = MappedFileReader::readFromStream(stream, filename, mtime, verbose, defaults, errorDescription, hash,
-                                                     size);
+  result_t result
+  = MappedFileReader::readFromStream(stream, filename, mtime, verbose, defaults, errorDescription, replace, hash, size);
   if (defaults) {
     string circuit = AttributedItem::pluck("circuit", defaults);
     if (!circuit.empty() && m_circuitData.find(circuit) == m_circuitData.end()) {
@@ -2134,7 +2228,7 @@ result_t MessageMap::readFromStream(istream* stream, const string& filename, con
 }
 
 result_t MessageMap::addFromFile(const string& filename, unsigned int lineNo, map<string, string>* row,
-    vector< map<string, string> >* subRows, string* errorDescription) {
+    vector< map<string, string> >* subRows, string* errorDescription, bool replace) {
   Condition* condition = NULL;
   string types = AttributedItem::pluck("type", row);
   result_t result = readConditions(filename, &types, errorDescription, &condition);
@@ -2188,7 +2282,7 @@ result_t MessageMap::addFromFile(const string& filename, unsigned int lineNo, ma
     }
     for (const auto message : messages) {
       if (result == RESULT_OK) {
-        result = add(true, message);
+        result = add(true, message, replace);
         if (result == RESULT_ERR_DUPLICATE_NAME) {
           *errorDescription = "invalid name";
         } else if (result == RESULT_ERR_DUPLICATE) {

src/lib/ebus/message.h

@@ -782,6 +782,18 @@ class MessagePriorityQueue
     }
     priority_queue<Message*, vector<Message*>, compareMessagePriority>::push(__x);
   }
+  /**
+   * Remove data from the queue.
+   * @param __x the element to remove.
+   */
+  void remove(const value_type& __x) {
+    for (vector<Message*>::iterator it = c.begin(); it != c.end(); it++) {
+      if (*it == __x) {
+        c.erase(it);
+        break;
+      }
+    }
+  }
 };
 
 
@@ -1233,10 +1245,17 @@ class MessageMap : public MappedFileReader {
    * Add a @a Message instance to this set.
    * @param message the @a Message instance to add.
    * @param storeByName whether to store the @a Message by name.
+   * @param replace whether to replace an already existing entry.
    * @return @a RESULT_OK on success, or an error code.
    * Note: the caller may not free the added instance on success.
    */
-  result_t add(bool storeByName, Message* message);
+  result_t add(bool storeByName, Message* message, bool replace = false);
+
+  /**
+   * Remove a previously added @a Message.
+   * @param message the @a Message to remove.
+   */
+  void remove(Message* message);
 
   // @copydoc
   result_t getFieldMap(const string& preferLanguage, vector<string>* row, string* errorDescription) const override;
@@ -1261,11 +1280,12 @@ class MessageMap : public MappedFileReader {
 
   // @copydoc
   result_t readFromStream(istream* stream, const string& filename, const time_t& mtime, bool verbose,
-      map<string, string>* defaults, string* errorDescription, size_t* hash = NULL, size_t* size = NULL) override;
+      map<string, string>* defaults, string* errorDescription, bool replace = false, size_t* hash = NULL,
+      size_t* size = NULL) override;
 
   // @copydoc
   result_t addFromFile(const string& filename, unsigned int lineNo, map<string, string>* row,
-      vector< map<string, string> >* subRows, string* errorDescription) override;
+      vector< map<string, string> >* subRows, string* errorDescription, bool replace) override;
 
   /**
    * Get the scan @a Message instance for the specified address.