Feat[Storagetool]: Add support of queueOp and journalOp records

src/applications/bmqstoragetool/README.md

@@ -2,8 +2,8 @@ BMQStorageTool
 ==============
 
 BMQStorageTool is a command-line tool for analyzing of BlazingMQ Broker storage
-files. It allows to search messages in `journal` file with set of different 
-filters and output found message GUIDs or message detail information.
+files. It allows to search records in `journal` file with set of different 
+filters and output found results in the short or detail form.
 As input, a `journal` file (*.bmq_journal) is *always* required. To dump 
 payload, `data` file (*.bmq_data) is required. To filter by queue Uri, cluster
 state ledger (CSL) file (*.bmq_csl) is required.
@@ -13,7 +13,8 @@ the project. From the command-line, there are a few options you can use when
 invoking the tool.
 
 ```bash
-Usage:   bmqstoragetool [--journal-path <journal path>]
+Usage:   bmqstoragetool [-r|record-type <record type>]*
+                        [--journal-path <journal path>]
                         [--journal-file <journal file>]
                         [--data-file <data file>]
                         [--csl-file <csl file>]
@@ -37,6 +38,8 @@ Usage:   bmqstoragetool [--journal-path <journal path>]
                         [--summary]
                         [-h|help]
 Where:
+  -r | --record-type          <record type>
+          record type to search {message|queue-op|journal-op} (default: message)
        --journal-path         <pattern>
           '*'-ended file path pattern, where the tool will try to find journal
           and data files
@@ -104,6 +107,15 @@ Example:
 bmqstoragetool --journal-file=<path>
 ```
 
+Search and otput all queueOp/journalOp records or all records in journal file
+--------------------------------------------------
+Example:
+```bash
+bmqstoragetool --journal-file=<path> --record-type=queue-op
+bmqstoragetool --journal-file=<path> --record-type=journal-op
+bmqstoragetool --journal-file=<path> --record-type=journal-op --record-type=queue-op --record-type=message
+```
+
 Search and otput all messages details in journal file
 -----------------------------------------------------
 Example:

src/applications/bmqstoragetool/bmqstoragetool.m.cpp

@@ -24,6 +24,7 @@
 
 // BDE
 #include <balcl_commandline.h>
+#include <bdls_filesystemutil.h>
 #include <bsl_iostream.h>
 #include <bslma_managedptr.h>
 
@@ -38,6 +39,13 @@ static bool parseArgs(CommandLineArguments& arguments,
     bool showHelp = false;
 
     balcl::OptionInfo specTable[] = {
+        {"r|record-type",
+         "record type",
+         "record type to search {message|queue-op|journal-op}",
+         balcl::TypeInfo(&arguments.d_recordType,
+                         CommandLineArguments::isValidRecordType),
+         balcl::OccurrenceInfo(bsl::vector<bsl::string>(
+             {CommandLineArguments::k_MESSAGE_TYPE}))},
         {"journal-path",
          "journal path",
          "'*'-ended file path pattern, where the tool will try to find "
@@ -47,17 +55,20 @@ static bool parseArgs(CommandLineArguments& arguments,
         {"journal-file",
          "journal file",
          "path to a .bmq_journal file",
-         balcl::TypeInfo(&arguments.d_journalFile),
+         balcl::TypeInfo(&arguments.d_journalFile,
+                         CommandLineArguments::isValidFileName),
          balcl::OccurrenceInfo::e_OPTIONAL},
         {"data-file",
          "data file",
          "path to a .bmq_data file",
-         balcl::TypeInfo(&arguments.d_dataFile),
+         balcl::TypeInfo(&arguments.d_dataFile,
+                         CommandLineArguments::isValidFileName),
          balcl::OccurrenceInfo::e_OPTIONAL},
         {"csl-file",
          "csl file",
          "path to a .bmq_csl file",
-         balcl::TypeInfo(&arguments.d_cslFile),
+         balcl::TypeInfo(&arguments.d_cslFile,
+                         CommandLineArguments::isValidFileName),
          balcl::OccurrenceInfo::e_OPTIONAL},
         {"guid",
          "guid",
@@ -148,8 +159,7 @@ static bool parseArgs(CommandLineArguments& arguments,
          balcl::OccurrenceInfo(1024)},
         {"summary",
          "summary",
-         "summary of all matching messages (number of outstanding messages "
-         "and other statistics)",
+         "summary of all matching records",
          balcl::TypeInfo(&arguments.d_summary),
          balcl::OccurrenceInfo::e_OPTIONAL},
         {"h|help",

src/applications/bmqstoragetool/m_bmqstoragetool_filters.cpp

@@ -51,16 +51,20 @@ Filters::Filters(const bsl::vector<bsl::string>& queueKeys,
     }
 }
 
-bool Filters::apply(const mqbs::MessageRecord& record,
-                    bsls::Types::Uint64        offset) const
+bool Filters::apply(const mqbs::RecordHeader& recordHeader,
+                    bsls::Types::Uint64       recordOffset,
+                    const mqbu::StorageKey&   queueKey,
+                    bool*                     highBoundReached_p) const
 {
+    if (highBoundReached_p) {
+        *highBoundReached_p = false;  // by default
+    }
+
     // Apply `queue key` filter
-    if (!d_queueKeys.empty()) {
+    if (queueKey != mqbu::StorageKey::k_NULL_KEY && !d_queueKeys.empty()) {
         // Match by queueKey
-        bsl::unordered_set<mqbu::StorageKey>::const_iterator it = bsl::find(
-            d_queueKeys.cbegin(),
-            d_queueKeys.cend(),
-            record.queueKey());
+        bsl::unordered_set<mqbu::StorageKey>::const_iterator it =
+            bsl::find(d_queueKeys.cbegin(), d_queueKeys.cend(), queueKey);
         if (it == d_queueKeys.cend()) {
             // Not matched
             return false;  // RETURN
@@ -71,32 +75,41 @@ bool Filters::apply(const mqbs::MessageRecord& record,
     bsls::Types::Uint64 value, valueGt, valueLt;
     switch (d_range.d_type) {
     case Parameters::Range::e_TIMESTAMP:
-        value   = record.header().timestamp();
+        value   = recordHeader.timestamp();
         valueGt = d_range.d_timestampGt;
         valueLt = d_range.d_timestampLt;
         break;
     case Parameters::Range::e_OFFSET:
-        value   = offset;
+        value   = recordOffset;
         valueGt = d_range.d_offsetGt;
         valueLt = d_range.d_offsetLt;
         break;
     case Parameters::Range::e_SEQUENCE_NUM: {
-        CompositeSequenceNumber seqNum(record.header().primaryLeaseId(),
-                                       record.header().sequenceNumber());
+        CompositeSequenceNumber seqNum(recordHeader.primaryLeaseId(),
+                                       recordHeader.sequenceNumber());
+        const bool greaterOrEqualToHigherBound = d_range.d_seqNumLt.isSet() &&
+                                                 d_range.d_seqNumLt <= seqNum;
+        if (highBoundReached_p && greaterOrEqualToHigherBound) {
+            *highBoundReached_p = true;
+        }
+
         return !(
             (d_range.d_seqNumGt.isSet() && seqNum <= d_range.d_seqNumGt) ||
-            (d_range.d_seqNumLt.isSet() &&
-             d_range.d_seqNumLt <= seqNum));  // RETURN
+            greaterOrEqualToHigherBound);  // RETURN
     } break;
     default:
         // No range filter defined
         return true;  // RETURN
     }
-    if ((valueGt > 0 && value <= valueGt) ||
-        (valueLt > 0 && value >= valueLt)) {
+    const bool greaterOrEqualToHigherBound = valueLt > 0 && value >= valueLt;
+    if ((valueGt > 0 && value <= valueGt) || greaterOrEqualToHigherBound) {
+        if (highBoundReached_p && greaterOrEqualToHigherBound) {
+            *highBoundReached_p = true;
+        }
         // Not inside range
         return false;  // RETURN
     }
+
     return true;
 }
 

src/applications/bmqstoragetool/m_bmqstoragetool_filters.h

@@ -27,6 +27,7 @@
 #include <m_bmqstoragetool_parameters.h>
 
 // MQB
+#include <mqbs_filestoreprotocol.h>
 #include <mqbu_storagekey.h>
 
 // BDE
@@ -57,10 +58,15 @@ class Filters {
 
     // ACCESSORS
 
-    /// Apply filters at specified 'record' and return true if all filters
-    /// are matched, false otherwise.
-    bool apply(const mqbs::MessageRecord& record,
-               bsls::Types::Uint64        offset) const;
+    /// Apply filters at the record with the specified `recordHeader`,
+    /// `recordOffset` and `queueKey`. Return true if all filters are matched,
+    /// false otherwise. If the specified `highBoundReached_p` pointer is
+    /// present, pointed value is set to true if higher bound value is reached,
+    /// false otherwise.
+    bool apply(const mqbs::RecordHeader& recordHeader,
+               bsls::Types::Uint64       recordOffset,
+               const mqbu::StorageKey&   queueKey,
+               bool*                     highBoundReached_p = 0) const;
 };
 
 }  // close package namespace

src/applications/bmqstoragetool/m_bmqstoragetool_journalfileprocessor.cpp

@@ -197,33 +197,70 @@ void JournalFileProcessor::process()
             needMoveToLowerBound = false;
         }
 
-        // MessageRecord
-        if (iter->recordType() == mqbs::RecordType::e_MESSAGE) {
-            const mqbs::MessageRecord& record = iter->asMessageRecord();
+        // Process Message records
+        if (d_parameters->d_processRecordTypes.d_message) {
+            // MessageRecord
+            if (iter->recordType() == mqbs::RecordType::e_MESSAGE) {
+                const mqbs::MessageRecord& record = iter->asMessageRecord();
 
-            // Apply filters
-            if (filters.apply(record, iter->recordOffset())) {
-                stopSearch = d_searchResult_p->processMessageRecord(
+                // Apply filters
+                if (filters.apply(iter->recordHeader(),
+                                  iter->recordOffset(),
+                                  record.queueKey())) {
+                    stopSearch = d_searchResult_p->processMessageRecord(
+                        record,
+                        iter->recordIndex(),
+                        iter->recordOffset());
+                }
+            }
+            // ConfirmRecord
+            else if (iter->recordType() == mqbs::RecordType::e_CONFIRM) {
+                const mqbs::ConfirmRecord& record = iter->asConfirmRecord();
+                stopSearch = d_searchResult_p->processConfirmRecord(
+                    record,
+                    iter->recordIndex(),
+                    iter->recordOffset());
+            }
+            // DeletionRecord
+            else if (iter->recordType() == mqbs::RecordType::e_DELETION) {
+                const mqbs::DeletionRecord& record = iter->asDeletionRecord();
+                stopSearch = d_searchResult_p->processDeletionRecord(
                     record,
                     iter->recordIndex(),
                     iter->recordOffset());
             }
         }
-        // ConfirmRecord
-        else if (iter->recordType() == mqbs::RecordType::e_CONFIRM) {
-            const mqbs::ConfirmRecord& record = iter->asConfirmRecord();
-            stopSearch = d_searchResult_p->processConfirmRecord(
-                record,
-                iter->recordIndex(),
-                iter->recordOffset());
+        // Process QueueOp record
+        if (d_parameters->d_processRecordTypes.d_queueOp &&
+            iter->recordType() == mqbs::RecordType::e_QUEUE_OP) {
+            const mqbs::QueueOpRecord& record = iter->asQueueOpRecord();
+
+            // Apply filters
+            if (filters.apply(iter->recordHeader(),
+                              iter->recordOffset(),
+                              record.queueKey(),
+                              &stopSearch)) {
+                stopSearch = d_searchResult_p->processQueueOpRecord(
+                    record,
+                    iter->recordIndex(),
+                    iter->recordOffset());
+            }
         }
-        // DeletionRecord
-        else if (iter->recordType() == mqbs::RecordType::e_DELETION) {
-            const mqbs::DeletionRecord& record = iter->asDeletionRecord();
-            stopSearch = d_searchResult_p->processDeletionRecord(
-                record,
-                iter->recordIndex(),
-                iter->recordOffset());
+        // Process JournalOp record
+        if (d_parameters->d_processRecordTypes.d_journalOp &&
+            iter->recordType() == mqbs::RecordType::e_JOURNAL_OP) {
+            const mqbs::JournalOpRecord& record = iter->asJournalOpRecord();
+
+            // Apply filters
+            if (filters.apply(iter->recordHeader(),
+                              iter->recordOffset(),
+                              mqbu::StorageKey::k_NULL_KEY,
+                              &stopSearch)) {
+                stopSearch = d_searchResult_p->processJournalOpRecord(
+                    record,
+                    iter->recordIndex(),
+                    iter->recordOffset());
+            }
         }
     }
 }