class_docstr: The IPython Controller Hub with 0MQ connections
  
  Parameters
  ==========
  loop: zmq IOLoop instance
  session: Session object
  <removed> context: zmq context for creating new connections (?)
  queue: ZMQStream for monitoring the command queue (SUB)
  query: ZMQStream for engine registration and client queries requests (ROUTER)
  heartbeat: HeartMonitor object checking the pulse of the engines
  notifier: ZMQStream for broadcasting engine registration changes (PUB)
  db: connection to db for out of memory logging of commands
              NotImplemented
  engine_info: dict of zmq connection information for engines to connect
              to the queues.
  client_info: dict of zmq connection information for engines to connect
              to the queues.
class_docstr_tok: ["The","IPython","Controller","Hub","with","0MQ","connections","Parameters","==========","loop",":","zmq","IOLoop","instance","session",":","Session","object","<","removed",">","context",":","zmq","context","for","creating","new","connections","(","?",")","queue",":","ZMQStream","for","monitoring","the","command","queue","(","SUB",")","query",":","ZMQStream","for","engine","registration","and","client","queries","requests","(","ROUTER",")","heartbeat",":","HeartMonitor","object","checking","the","pulse","of","the","engines","notifier",":","ZMQStream","for","broadcasting","engine","registration","changes","(","PUB",")","db",":","connection","to","db","for","out","of","memory","logging","of","commands","NotImplemented","engine_info",":","dict","of","zmq","connection","information","for","engines","to","connect","to","the","queues",".","client_info",":","dict","of","zmq","connection","information","for","engines","to","connect","to","the","queues","."]
docstr: Resubmit one or more tasks.
docstr_tok: ["Resubmit","one","or","more","tasks","."]
code: def resubmit_task(self, client_id, msg):
      """Resubmit one or more tasks."""
      parent = msg
  
      def finish(reply):
          self.session.send(
              self.query,
              "resubmit_reply",
              content=reply,
              ident=client_id,
              parent=parent,
          )
  
      content = msg["content"]
      msg_ids = content["msg_ids"]
      reply = dict(status="ok")
      try:
          records = self.db.find_records(
              {"msg_id": {"$in": msg_ids}},
              keys=["header", "content", "buffers"],
          )
      except Exception:
          self.log.error(
              "db::db error finding tasks to resubmit", exc_info=True
          )
          return finish(error.wrap_exception())
  
      # validate msg_ids
      found_ids = [rec["msg_id"] for rec in records]
      pending_ids = [
          msg_id for msg_id in found_ids if msg_id in self.pending
      ]
      if len(records) > len(msg_ids):
          try:
              raise RuntimeError(
                  "DB appears to be in an inconsistent state."
                  "More matching records were found than should exist"
              )
          except Exception:
              self.log.exception("Failed to resubmit task")
              return finish(error.wrap_exception())
      elif len(records) < len(msg_ids):
          missing = [m for m in msg_ids if m not in found_ids]
          try:
              raise KeyError("No such msg(s): %r" % missing)
          except KeyError:
              self.log.exception("Failed to resubmit task")
              return finish(error.wrap_exception())
      elif pending_ids:
          pass
          # no need to raise on resubmit of pending task, now that we
          # resubmit under new ID, but do we want to raise anyway?
          # msg_id = invalid_ids[0]
          # try:
          #     raise ValueError("Task(s) %r appears to be inflight" % )
          # except Exception:
          #     return finish(error.wrap_exception())
  
      # mapping of original IDs to resubmitted IDs
      resubmitted = {}
  
      # send the messages
      for rec in records:
          header = rec["header"]
          msg = self.session.msg(header["msg_type"], parent=header)
          msg_id = msg["msg_id"]
          msg["content"] = rec["content"]
  
          # use the old header, but update msg_id and timestamp
          fresh = msg["header"]
          header["msg_id"] = fresh["msg_id"]
          header["date"] = fresh["date"]
          msg["header"] = header
  
          self.session.send(self.resubmit, msg, buffers=rec["buffers"])
  
          resubmitted[rec["msg_id"]] = msg_id
          self.pending.add(msg_id)
          msg["buffers"] = rec["buffers"]
          try:
              self.db.add_record(msg_id, init_record(msg))
          except Exception:
              self.log.error(
                  "db::DB Error updating record: %s",
                  msg_id,
                  exc_info=True,
              )
              return finish(error.wrap_exception())
  
      finish(dict(status="ok", resubmitted=resubmitted))
  
      # store the new IDs in the Task DB
      for msg_id, resubmit_id in resubmitted.items():
          try:
              self.db.update_record(
                  msg_id, {"resubmitted": resubmit_id}
              )
          except Exception:
              self.log.error(
                  "db::DB Error updating record: %s",
                  msg_id,
                  exc_info=True,
              )
  
imports: import inspect
  import json
  import os
  import sys
  import time
  from collections import deque
  from datetime import datetime
  import zmq
  from jupyter_client.jsonutil import parse_date
  from jupyter_client.session import Session
  from tornado import ioloop
  from traitlets import Any, Bytes, Dict, Float, HasTraits, Instance, Integer, Set, Unicode, default
  from traitlets.config import LoggingConfigurable
  from zmq.eventloop.zmqstream import ZMQStream
  from ipyparallel import error, util
  from ..util import extract_dates
  from .heartmonitor import HeartMonitor
label_desc: Use image node_id 33 for calling the Hub obj's underlying member method code with example usage: obj.resubmit_task(client_id, msg) and returns: finish, finish, finish, finish