def append_sim_table(self, coinc_file, sim_file):
    """
    Append injection SimInspiral Table to the event
    coinc file object.
    """
    # init a new sim inspiral table
    this_sim_table = lsctables.SimInspiralTable.get_table(sim_file)
    coinc_file.childNodes[-1].appendChild(this_sim_table)

    return coinc_file

def construct_event_ouput(self, xmldoc, event, injection=None, key=None):
    """
    Construct output message payload to be sent
    to Kafka.

    Parameters
    ----------
    xmldoc (ligolw document)
        event coinc file object
    event (dict)
        event json packet
    injection (dict)
        injection json packet
    key (str)
        optional tag used in writing files
        to disk, default is None.
    """
    filename = (
        f'coinc-{int(event["time"])}.xml'
        if not key
        else f'{key}-coinc-{int(event["time"])}.xml'
    )

    coinc = event["coinc"]
    coincrow = lsctables.CoincInspiralTable.get_table(coinc)[0]

    if injection:
        simrow = lsctables.SimInspiralTable.get_table(coinc)[0]
        onifos = injection["ifos"]
        time = simrow.geocent_end_time
        time_ns = simrow.geocent_end_time_ns
    else:
        simrow = None
        time = event["time"]
        time_ns = 0
        onifos = ""

    ligolw_utils.write_filename(
        xmldoc, os.path.join("coincs", filename), verbose=self.verbose
    )
    coinc_msg = io.BytesIO()
    ligolw_utils.write_fileobj(xmldoc, coinc_msg)

    output = {
        "time": time,
        "time_ns": time_ns,
        "snr": coincrow.snr,
        "far": coincrow.combined_far,
        "p_astro": event["p_astro"],
        "coinc": coinc_msg.getvalue().decode(),
        "latency": event["latency"],
        "uid": event["uid"],
        "onIFOs": onifos,
        "pipeline": event["pipeline"],
        "snr_optimized": event["snr_optimized"],
    }

    return output

def get_preferred_param(self, coinc):
    """
    Parse coinc file object for the preferred
    parameter (either ifar or snr).
    """
    # IFAR
    ifar = 1.0 / coinc.getColumnByName("combined_far")[0]
    # SNR
    snr = coinc.getColumnByName("snr")[0]
    # get preferred param value for this event
    if self.preferred_param == "ifar":
        return (ifar, snr)
    elif self.preferred_param == "snr":
        return (snr, ifar)
    else:
        raise NotImplementedError

def process_events(self, datasource):
    """
    For each event in the event_msgs deque, find the nearest injection
    in inj_msgs within +/- delta_t (1 second) of the event coalescence
    time. When an association is made, check to see if its better than
    any previous event found. If so, add the sim inspiral table from
    injection to the event's coinc xml and send a message to the
    testsuite.events topic and remove the processed event from the
    deque.
    """
    events_copy = copy.copy(self.event_msgs[datasource])
    injections = self.inj_msgs[datasource]

    for event in events_copy:
        event_time = event["time"]
        nearest_inj = utils.find_nearest_msg(injections, event_time)

        # if no associated injection was found, continue
        if not nearest_inj:
            logging.info(f"No injection found for event at {event_time}")
            continue

        inj_idx = self.inj_msgs[datasource].index(nearest_inj)
        inj_time = nearest_inj["time"]
        sim_file = nearest_inj["sim"]
        prev_preferred_event = nearest_inj["preferred_event"]
        coinc_file = event["coinc"]
        this_coinc = lsctables.CoincInspiralTable.get_table(coinc_file)
        val = self.get_preferred_param(this_coinc)

        # if this is the first event found or
        # this event is better than the previous,
        # send update event.
        # Note: this requires that aggregate by
        # "latest" works the way we would hope
        if not prev_preferred_event or val > prev_preferred_event:
            # update preferred event for this injection
            injections[inj_idx].update({"preferred_event": val})

            # proceed with sending event
            # add sim table to coinc file and write to disk
            logging.info(
                f"Sending event with {self.preferred_param} = {val} " +
                f"for injection at time {inj_time}"
            )
            newxmldoc = self.append_sim_table(coinc_file, sim_file)
            output = self.construct_event_ouput(
                                                newxmldoc,
                                                event, nearest_inj
            )

            topic = f"{datasource}.{self.tag}.testsuite.events"
            self.client.write(topic, output)
            logging.info(f'Sent msg to: {topic}')
        else:
            logging.debug(
                f"This event {val} not preferred over previous: " +
                f"{prev_preferred_event}, skipping."
            )

def process_stale_msgs(self, datasource):
    """
    process old events or injections that have passed the timeout
    and send a message with the necessary info
    this is necessary in the case that:
        * we receive an event from the search which is not
        associated with an injection, ie a glitch or real gw
        candidate.
        * there is an injection for which we never receive
        an associated event from the search. ie the injection
        was not recovered at even the GDB far threshold.
    """
    stale_events = self.stale_msgs(self.event_msgs[datasource])
    for stale_event in stale_events:
        stale_time = stale_event["time"]
        nearest_inj = utils.find_nearest_msg(
                self.inj_msgs[datasource], stale_time
        )

        if not nearest_inj:
            logging.info(
                f'Found stale event from time {stale_time} with no ' +
                'nearby injection. Counting as terrestrial and ' +
                'removing from the deque.'
            )

            output = self.construct_event_ouput(
                                                stale_event["coinc"],
                                                stale_event
            )

            topic = f"{datasource}.{self.tag}.testsuite.terr_events"
            self.client.write(topic, output)

            logging.info(f'Sent msg to: {topic}')

            # finally remove event from the deque
            self.event_msgs[datasource].remove(stale_event)

        else:
            # this is not a terrestrial event because it has
            # a nearby injection. But since it is passed the time
            # out, we can remove it and its associated injection
            # at the same time.
            logging.debug(
                f'Removing event and injection from {stale_time} ' +
                'from the deques after reaching timeout.')
            self.event_msgs[datasource].remove(stale_event)
            self.inj_msgs[datasource].remove(nearest_inj)

    stale_injections = self.stale_msgs(self.inj_msgs[datasource])
    for stale_inj in stale_injections:
        stale_time = stale_inj["time"]
        if not stale_inj["preferred_event"]:
            sim_inspiral = stale_inj["sim"]
            logging.info(
                f'Sending {datasource} missed injection msg ' +
                f'for injection {stale_time}'
            )
            simrow = lsctables.SimInspiralTable.get_table(sim_inspiral)[0]
            newxmldoc = ligolw.Document()
            sim_msg = self.write_sim_file(sim_inspiral, newxmldoc)

            output = {
                "time": simrow.geocent_end_time,
                "time_ns": simrow.geocent_end_time_ns,
                "sim": sim_msg.getvalue().decode(),
                "onIFOs": stale_inj["ifos"],
            }

            topic = f"{datasource}.{self.tag}.testsuite.missed_inj"
            self.client.write(topic, output)
            logging.info(f"Sent msg to: {topic}")
            newxmldoc.unlink()

            # finally remove this injection from the deque
            self.inj_msgs[datasource].remove(stale_inj)

        else:
            # this is not a missed injection because a previous
            # event has been found associated with it. But since
            # it is passed the time out, we can remove it and its
            # associated event at the same time.
            nearest_event = utils.find_nearest_msg(
                    self.event_msgs[datasource], stale_time
            )
            logging.debug(
                f'Removing event and injection from {stale_time} ' +
                'from the deques after reaching timeout.')
            self.event_msgs[datasource].remove(nearest_event)
            self.inj_msgs[datasource].remove(stale_inj)

def stale_msgs(self, deque):
    """
    Determine if there are stale messages
    in the queue to be removed.
    """
    stale = []
    now = float(GPSTimeNow())
    for item in deque:
        if now - item["time"] >= self.timeout:
            stale.append(item)
    return stale

def start(self):
    """
    Start job service.
    """
    # start up
    logging.info("Starting up...")
    self.app.start()

def write_sim_file(self, sim, xmldoc):
    """
    Write a ligolw file object including the
    injection SimInspiral Table
    """
    # open a new xml doc
    sim_msg = io.BytesIO()
    ligolw_elem = xmldoc.appendChild(ligolw.LIGO_LW())

    output_simtable = ligolw_elem.appendChild(
        lsctables.New(lsctables.SimInspiralTable)
    )
    this_sim_table = lsctables.SimInspiralTable.get_table(sim)
    output_simtable.extend(this_sim_table)
    ligolw_utils.write_fileobj(xmldoc, sim_msg)

    return sim_msg