docs: def run(self):
          """Redirects messages until a shutdown message is received."""
          while True:
              if not self.task_socket.poll(-1):
                  continue
  
              msg = self.task_socket.recv_multipart()
              msg_type = msg[1]
  
              if self.debug:
                  self.stats.append((time.time(),
                                     msg_type,
                                     len(self.unassigned_tasks),
                                     len(self.available_workers)))
                  if time.time() - self.lastDebugTs > TIME_BETWEEN_PARTIALDEBUG:
                      self.writeDebug("debug/partial-{0}".format(
                          round(time.time(), -1)
                      ))
                      self.lastDebugTs = time.time()
  
              # New task inbound
              if msg_type == TASK:
                  task_id = msg[2]
                  task = msg[3]
                  self.logger.debug("Received task {0}".format(task_id))
                  try:
                      address = self.available_workers.popleft()
                  except IndexError:
                      self.unassigned_tasks.append((task_id, task))
                  else:
                      self.logger.debug("Sent {0}".format(task_id))
                      self.task_socket.send_multipart([address, TASK, task])
                      self.assigned_tasks[address].add(task_id)
  
              # Request for task
              elif msg_type == REQUEST:
                  address = msg[0]
                  try:
                      task_id, task = self.unassigned_tasks.popleft()
                  except IndexError:
                      self.available_workers.append(address)
                  else:
                      self.logger.debug("Sent {0}".format(task_id))
                      self.task_socket.send_multipart([address, TASK, task])
                      self.assigned_tasks[address].add(task_id)
  
              # A task status request is requested
              elif msg_type == STATUS_REQ:
                  self.pruneAssignedTasks()
                  address = msg[0]
                  task_id = msg[2]
  
                  if any(task_id in x for x in self.assigned_tasks.values()):
                      status = STATUS_GIVEN
                  elif task_id in (x[0] for x in self.unassigned_tasks):
                      status = STATUS_HERE
                  else:
                      status = STATUS_NONE
  
                  self.task_socket.send_multipart([
                      address, STATUS_ANS, task_id, status
                  ])
  
              # A task status set (task done) is received
              elif msg_type == STATUS_DONE:
                  address = msg[0]
                  task_id = msg[2]
  
                  try:
                      self.assigned_tasks[address].discard(task_id)
                  except KeyError:
                      pass
  
              elif msg_type == STATUS_UPDATE:
                  address = msg[0]
                  try:
                      tasks_ids = pickle.loads(msg[2])
                  except:
                      self.logger.error("Could not unpickle status update message.")
                  else:
                      self.assigned_tasks[address] = tasks_ids
                      self.status_times[address] = time.time()
  
              # Answer needing delivery
              elif msg_type == REPLY:
                  self.logger.debug("Relaying")
                  destination = msg[-1]
                  origin = msg[0]
                  self.task_socket.send_multipart([destination] + msg[1:] + [origin])
  
              # Shared variable to distribute
              elif msg_type == VARIABLE:
                  address = msg[4]
                  value = msg[3]
                  key = msg[2]
                  self.shared_variables[address].update(
                      {key: value},
                  )
                  self.info_socket.send_multipart([VARIABLE,
                                                  key,
                                                  value,
                                                  address])
  
              # Initialize the variables of a new worker
              elif msg_type == INIT:
                  address = msg[0]
                  try:
                      self.processConfig(pickle.loads(msg[2]))
                  except pickle.PickleError:
                      continue
                  self.task_socket.send_multipart([
                      address,
                      pickle.dumps(self.config,
                                   pickle.HIGHEST_PROTOCOL),
                      pickle.dumps(self.shared_variables,
                                   pickle.HIGHEST_PROTOCOL),
                  ])
  
                  self.task_socket.send_multipart([
                      address,
                      pickle.dumps(self.cluster_available,
                                   pickle.HIGHEST_PROTOCOL),
                  ])
  
              # Add a given broker to its fellow list
              elif msg_type == CONNECT:
                  try:
                      connect_brokers = pickle.loads(msg[2])
                  except pickle.PickleError:
                      self.logger.error("Could not understand CONNECT message.")
                      continue
                  self.logger.info("Connecting to other brokers...")
                  self.addBrokerList(connect_brokers)
  
              # Shutdown of this broker was requested
              elif msg_type == SHUTDOWN:
                  self.logger.debug("SHUTDOWN command received.")
                  self.shutdown()
                  break