code: async def get(self, *, encoding=None, decoder=None):
          
          # TODO: add note about raised exception and end marker.
          #   Flow before ClosableQueue:
          #   - ch.get() -> message
          #   - ch.close() -> ch.put(None)
          #   - ch.get() -> None
          #   - ch.get() -> ChannelClosedError
          #   Current flow:
          #   - ch.get() -> message
          #   - ch.close() -> ch._closed = True
          #   - ch.get() -> ChannelClosedError
          assert decoder is None or callable(decoder), decoder
          if self._queue.exhausted:
              raise ChannelClosedError()
          obj = await self._queue.get()
          if obj is EndOfStream:
              return
          ch, msg = obj
          if ch.is_pattern:
              dest_ch, msg = msg
          if encoding is not None:
              msg = msg.decode(encoding)
          if decoder is not None:
              msg = decoder(msg)
          if ch.is_pattern:
              return ch, (dest_ch, msg)
          return ch, msg
docs: Wait for and return pub/sub message from one of channels.
  
          Return value is either:
  
          * tuple of two elements: channel & message;
  
          * tuple of three elements: pattern channel, (target channel & message);
  
          * or None in case Receiver is not active or has just been stopped.
  
          :raises aioredis.ChannelClosedError: If listener is stopped
              and all messages have been received.