tests/data/plugins/sink.ctf.fs/assume-single-trace/bt_plugin_foo.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2020 EfficiOS Inc.
   4 #
   5
   6 import bt2
   7
   8
   9 class TheSourceIterator(bt2._UserMessageIterator):
  10     def __init__(self, config, port):
  11         tc, sc, ec = port.user_data
  12
  13         trace = tc()
  14         stream = trace.create_stream(sc, name="the-stream")
  15
  16         self._msgs = [
  17             self._create_stream_beginning_message(stream),
  18             self._create_event_message(ec, stream),
  19             self._create_stream_end_message(stream),
  20         ]
  21
  22     def __next__(self):
  23         if len(self._msgs) == 0:
  24             raise StopIteration
  25
  26         return self._msgs.pop(0)
  27
  28
  29 @bt2.plugin_component_class
  30 class TheSource(bt2._UserSourceComponent, message_iterator_class=TheSourceIterator):
  31     def __init__(self, config, params, obj):
  32         tc = self._create_trace_class()
  33         sc = tc.create_stream_class()
  34         ec = sc.create_event_class(name="the-event")
  35         self._add_output_port("out", user_data=(tc, sc, ec))
  36
  37
  38 bt2.register_plugin(__name__, "foo")