code: def publish(self, message):
          
          # Coerce the type, just in case.
          if not isinstance(message, types.PubsubMessage):
              message = types.PubsubMessage(**message)
  
          future = None
  
          with self._state_lock:
              if not self.will_accept(message):
                  return future
  
              new_size = self._size + message.ByteSize()
              new_count = len(self._messages) + 1
              overflow = (
                  new_size > self.settings.max_bytes
                  or new_count >= self._settings.max_messages
              )
  
              if not self._messages or not overflow:
  
                  # Store the actual message in the batch's message queue.
                  self._messages.append(message)
                  self._size = new_size
  
                  # Track the future on this batch (so that the result of the
                  # future can be set).
                  future = futures.Future(completed=threading.Event())
                  self._futures.append(future)
  
          # Try to commit, but it must be **without** the lock held, since
          # ``commit()`` will try to obtain the lock.
          if overflow:
              self.commit()
  
          return future
docs: Publish a single message.
  
          Add the given message to this object; this will cause it to be
          published once the batch either has enough messages or a sufficient
          period of time has elapsed.
  
          This method is called by :meth:`~.PublisherClient.publish`.
  
          Args:
              message (~.pubsub_v1.types.PubsubMessage): The Pub/Sub message.
  
          Returns:
              Optional[~google.api_core.future.Future]: An object conforming to
              the :class:`~concurrent.futures.Future` interface or :data:`None`.
              If :data:`None` is returned, that signals that the batch cannot
              accept a message.