The Listeners

A medium Python interview practice problem on DataDriven. Write the Python and run it against test cases, with instant feedback.

Domain
Python
Difficulty
medium
Seniority
L4
Asked in
Foursquare

Problem

Implement an EventEmitter with on(event, listener) to subscribe, emit(event, payload) that returns the payload once per listener currently on that event, and off(event, listener) that drops the first matching listener. Then implement event_broadcaster(op_names, op_args), which replays the parallel op sequence against one emitter and collects each result: None for the constructor, on, and off, and the payload list for emit.

Subscribers show up, listen, and sometimes leave.

Example

Input:

op_args = [[],["click","log_click"],["click","event1"],["click","log_click"],["click","event2"]], op_names = ["EventEmitter","on","emit","off","emit"]

Output:

[null,null,["event1"],null,[]]

Worked solution and explanation

What this really is

Strip the pub/sub costume and this is two data structures pretending to be one. The emitter is a keyed registry: a dict of event to an ordered list of listeners. The driver is an op-replay loop, the exact shape you see in LRU-cache and stack harnesses where the test scripts a sequence of method calls. The skill being probed is keeping them apart. Anyone can append to a list; the tell is whether your emit and off touch only the target event, and whether you resist letting the driver's bookkeeping leak into the emitter. Blur the two and emit for one event starts returning another event's listeners, or off strips the wrong subscription.

Break down the requirements

Step 1: Keyed registration with first-match removal

Inside EventEmitter, keep self.listeners as a dict[str, list]. on(event, listener) appends listener to self.listeners.setdefault(event, []). off(event, listener) removes the first matching listener from that event's list, and does nothing if the list is empty or the listener isn't found. The keying is load-bearing: a single flat list would mix events, and a set would drop the duplicate ordering that off depends on.

Step 2: emit returns one payload per active listener

emit(event, payload) returns [payload] * len(self.listeners.get(event, [])). The length is the number of listeners on that event at the moment of emission, so a duplicate subscription counts twice. Listeners are opaque identifiers, so nothing is actually invoked; the contract is simply to deliver the payload once per current listener, and an unseen or emptied event returns [].

Step 3: Driver loop over the parallel op arrays

event_broadcaster(op_names, op_args) walks the two parallel lists in lockstep. Index 0 is always 'EventEmitter' and constructs the instance (result None). For each later op, dispatch on the name and append the result: on and off produce None, emit produces the payload list. The driver holds no event logic of its own.

The solution

EventEmitter plus a driver that replays `op_names` against `op_args`
class EventEmitter:
    def __init__(self) -> None:
        self.listeners: dict[str, list] = {}

    def on(self, event, listener):
        self.listeners.setdefault(event, []).append(listener)

    def emit(self, event, payload):
        return [payload] * len(self.listeners.get(event, []))

    def off(self, event, listener):
        bucket = self.listeners.get(event)
        if not bucket:
            return
        try:
            bucket.remove(listener)
        except ValueError:
            pass


def event_broadcaster(op_names: list[str], op_args: list[list]) -> list:
    emitter = None
    out = []
    for name, args in zip(op_names, op_args):
        if name == 'EventEmitter':
            emitter = EventEmitter()
            out.append(None)
        elif name == 'on':
            emitter.on(*args)
            out.append(None)
        elif name == 'emit':
            out.append(emitter.emit(*args))
        elif name == 'off':
            emitter.off(*args)
            out.append(None)
    return out

Common follow-up questions