code: def run(self):
          
          # Create the thread pool.
          executor = concurrent.futures.ThreadPoolExecutor(
              max_workers=self._config['num_workers'])
  
          # Wait to ensure multiple senders can be synchronised.
          now = int(datetime.datetime.utcnow().timestamp())
          start_time = ((now + 29) // 30) * 30
          self._log.info('Waiting until {}'.format(
                         datetime.datetime.fromtimestamp(start_time)))
          while int(datetime.datetime.utcnow().timestamp()) < start_time:
              time.sleep(0.1)
  
          # Run the event loop.
          loop = asyncio.get_event_loop()
          try:
              loop.run_until_complete(self._run_loop(executor))
          except KeyboardInterrupt:
              pass
          finally:
              # Send the end of stream message to each stream.
              self._log.info('Shutting down, closing streams...')
              tasks = []
              for stream, item_group in self._streams:
                  tasks.append(stream.async_send_heap(item_group.get_end()))
              loop.run_until_complete(asyncio.gather(*tasks))
              self._log.info('... finished.')
              executor.shutdown()
docs: Starts the sender.