+ def sections(self):
+ return self._sections
+
+
+LttngMetadataConfigSection = namedtuple(
+ "LttngMetadataConfigSection", ["line", "timestamp", "is_empty"]
+)
+
+
+def _parse_metadata_sections_config(config_sections):
+ assert config_sections is not None
+ config_metadata_sections = []
+ append_empty_section = False
+ last_timestamp = 0
+ last_line = 0
+
+ for config_section in config_sections:
+ if config_section == "empty":
+ # Found a empty section marker. Actually append the section at the
+ # timestamp of the next concrete section.
+ append_empty_section = True
+ else:
+ assert type(config_section) is dict
+ line = config_section.get("line")
+ ts = config_section.get("timestamp")
+
+ # Sections' timestamps and lines must both be increasing.
+ assert ts > last_timestamp
+ last_timestamp = ts
+ assert line > last_line
+ last_line = line
+
+ if append_empty_section:
+ config_metadata_sections.append(
+ LttngMetadataConfigSection(line, ts, True)
+ )
+ append_empty_section = False
+
+ config_metadata_sections.append(LttngMetadataConfigSection(line, ts, False))
+
+ return config_metadata_sections
+
+
+def _split_metadata_sections(metadata_file_path, raw_config_sections):
+ assert isinstance(raw_config_sections, collections.abc.Sequence)
+
+ parsed_sections = _parse_metadata_sections_config(raw_config_sections)
+
+ sections = []
+ with open(metadata_file_path, "r") as metadata_file:
+ metadata_lines = [line for line in metadata_file]
+
+ config_metadata_sections_idx = 0
+ curr_metadata_section = bytearray()
+
+ for idx, line_content in enumerate(metadata_lines):
+ # Add one to the index to convert from the zero-indexing of the
+ # enumerate() function to the one-indexing used by humans when
+ # viewing a text file.
+ curr_line_number = idx + 1
+
+ # If there are no more sections, simply append the line.
+ if config_metadata_sections_idx + 1 >= len(parsed_sections):
+ curr_metadata_section += bytearray(line_content, "utf8")
+ continue
+
+ next_section_line_number = parsed_sections[
+ config_metadata_sections_idx + 1
+ ].line
+
+ # If the next section begins at the current line, create a
+ # section with the metadata we gathered so far.
+ if curr_line_number >= next_section_line_number:
+
+ # Flushing the metadata of the current section.
+ sections.append(
+ _LttngMetadataStreamSection(
+ parsed_sections[config_metadata_sections_idx].timestamp,
+ bytes(curr_metadata_section),
+ )
+ )