code: def connect_pub(self, timeout=None):
          '''
          Establish the publish connection
          '''
          if self.cpub:
              return True
  
          if self._run_io_loop_sync:
              with salt.utils.asynchronous.current_ioloop(self.io_loop):
                  if self.subscriber is None:
                      self.subscriber = salt.transport.ipc.IPCMessageSubscriber(
                      self.puburi,
                      io_loop=self.io_loop
                  )
                  try:
                      self.io_loop.run_sync(
                          lambda: self.subscriber.connect(timeout=timeout))
                      self.cpub = True
                  except Exception:
                      pass
          else:
              if self.subscriber is None:
                  self.subscriber = salt.transport.ipc.IPCMessageSubscriber(
                  self.puburi,
                  io_loop=self.io_loop
              )
  
              # For the asynchronous case, the connect will be defered to when
              # set_event_handler() is invoked.
              self.cpub = True
          return self.cpuf connect_pub(self, timeout=None):
          '''
          Establish the publish connection
          '''
          if self.cpub:
              return True
  
          if self._run_io_loop_sync:
              with salt.utils.asynchronous.current_ioloop(self.io_loop):
                  if self.subscriber is None:
                      self.subscriber = salt.transport.ipc.IPCMessageSubscriber(
                      self.puburi,
                      io_loop=self.io_loop
                  )
                  try:
                      self.io_loop.run_sync(
                          lambda: self.subscriber.connect(timeout=timeout))
                      self.cpub = True
                  except Exception:
                      pass
          else:
              if self.subscriber is None:
                  self.subscriber = salt.transport.ipc.IPCMessageSubscriber(
                  self.puburi,
                  io_loop=self.io_loop
              )
  
              # For the asynchronous case, the connect will be defered to when
              # set_event_handler() is invoked.
              self.cpub = True
          return self.cpub
docs: Establish the publish connection