code: async def run(self, **params):
          
          if self.json_stream:
              warnings.warn("already running", RuntimeWarning, stackelevel=2)
              return
          forced_params = {"stream": True}
          params = ChainMap(forced_params, params)
          try:
              # timeout has to be set to 0, None is not passed
              # Otherwise after 5 minutes the client
              # will close the connection
              # http://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.ClientSession.request
              response = await self.docker._query(
                  "events", method="GET", params=params, timeout=0
              )
              self.json_stream = await json_stream_result(
                  response, self._transform_event, human_bool(params["stream"])
              )
              try:
                  async for data in self.json_stream:
                      await self.channel.publish(data)
              finally:
                  if self.json_stream is not None:
                      await self.json_stream._close()
                  self.json_stream = None
          finally:
              # signal termination to subscribers
              await self.channel.publish(None)
docs: Query the events endpoint of the Docker daemon.
  
          Publish messages inside the asyncio queue.