Observer and Event-Driven Design

package main

import (
	"errors"
	"fmt"
)

type Event struct {
	OrderID  int
	Customer string
	Amount   int // taka
}

type Handler func(e Event) error

type subscription struct {
	id   int
	name string
	fn   Handler
}

// Bus is an in-process event bus: publishers and subscribers only know the bus and the topic.
type Bus struct {
	subs   map[string][]subscription
	nextID int
}

func NewBus() *Bus { return &Bus{subs: map[string][]subscription{}} }

func (b *Bus) Subscribe(topic, name string, fn Handler) int {
	b.nextID++
	b.subs[topic] = append(b.subs[topic], subscription{b.nextID, name, fn})
	return b.nextID
}

func (b *Bus) Unsubscribe(id int) {
	for topic, list := range b.subs {
		kept := list[:0:0]
		for _, s := range list {
			if s.id != id {
				kept = append(kept, s)
			}
		}
		b.subs[topic] = kept
	}
}

// Publish calls every subscriber in order. One failing subscriber doesn't stop the others.
func (b *Bus) Publish(topic string, e Event) {
	fmt.Printf("publish %s #%d\n", topic, e.OrderID)
	for _, s := range b.subs[topic] {
		if err := s.fn(e); err != nil {
			fmt.Printf("  %-9s : failed: %v\n", s.name, err)
		}
	}
}

func main() {
	bus := NewBus()

	// The order service only publishes. It doesn't know who listens.
	bus.Subscribe("order.placed", "sms", func(e Event) error {
		if e.OrderID == 503 {
			return errors.New("gateway down")
		}
		fmt.Printf("  %-9s : sent \"order #%d confirmed\" to %s\n", "sms", e.OrderID, e.Customer)
		return nil
	})

	points := map[string]int{}
	loyalty := func(sign int) Handler {
		return func(e Event) error {
			p := sign * (e.Amount / 100) // 1 point per 100 taka
			points[e.Customer] += p
			fmt.Printf("  %-9s : %s %+d points (total %d)\n", "loyalty", e.Customer, p, points[e.Customer])
			return nil
		}
	}
	bus.Subscribe("order.placed", "loyalty", loyalty(+1))
	bus.Subscribe("order.cancelled", "loyalty", loyalty(-1))

	analytics := bus.Subscribe("order.placed", "analytics", func(e Event) error {
		fmt.Printf("  %-9s : recorded #%d\n", "analytics", e.OrderID)
		return nil
	})

	bus.Publish("order.placed", Event{501, "Rahim", 720})
	bus.Publish("order.placed", Event{502, "Nadia", 1500})

	bus.Unsubscribe(analytics)
	fmt.Println("analytics unsubscribed")

	bus.Publish("order.placed", Event{503, "Rahim", 300})
	bus.Publish("order.cancelled", Event{502, "Nadia", 1500})
}

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class Event:
    order_id: int
    customer: str
    amount: int  # taka


Handler = Callable[[Event], None]


class Bus:
    """An in-process event bus: publishers and subscribers only know the bus and the topic."""

    def __init__(self) -> None:
        self._subs: dict[str, list[tuple[int, str, Handler]]] = {}
        self._next_id = 0

    def subscribe(self, topic: str, name: str, fn: Handler) -> int:
        self._next_id += 1
        self._subs.setdefault(topic, []).append((self._next_id, name, fn))
        return self._next_id

    def unsubscribe(self, sub_id: int) -> None:
        for topic, subs in self._subs.items():
            self._subs[topic] = [s for s in subs if s[0] != sub_id]

    def publish(self, topic: str, e: Event) -> None:
        """Calls every subscriber in order. One failing subscriber doesn't stop the others."""
        print(f"publish {topic} #{e.order_id}")
        for _, name, fn in self._subs.get(topic, []):
            try:
                fn(e)
            except Exception as err:
                print(f"  {name:<9} : failed: {err}")


bus = Bus()


# The order service only publishes. It doesn't know who listens.
def sms(e: Event) -> None:
    if e.order_id == 503:
        raise RuntimeError("gateway down")
    print(f'  {"sms":<9} : sent "order #{e.order_id} confirmed" to {e.customer}')


bus.subscribe("order.placed", "sms", sms)

points: dict[str, int] = {}


def loyalty(sign: int) -> Handler:
    def handle(e: Event) -> None:
        p = sign * (e.amount // 100)  # 1 point per 100 taka
        points[e.customer] = points.get(e.customer, 0) + p
        print(f"  {'loyalty':<9} : {e.customer} {p:+d} points (total {points[e.customer]})")
    return handle


bus.subscribe("order.placed", "loyalty", loyalty(+1))
bus.subscribe("order.cancelled", "loyalty", loyalty(-1))

analytics = bus.subscribe("order.placed", "analytics",
                          lambda e: print(f"  {'analytics':<9} : recorded #{e.order_id}"))

bus.publish("order.placed", Event(501, "Rahim", 720))
bus.publish("order.placed", Event(502, "Nadia", 1500))

bus.unsubscribe(analytics)
print("analytics unsubscribed")

bus.publish("order.placed", Event(503, "Rahim", 300))
bus.publish("order.cancelled", Event(502, "Nadia", 1500))

interface OrderEvent {
  readonly orderId: number;
  readonly customer: string;
  readonly amount: number; // taka
}

type Handler = (e: OrderEvent) => void;

// An in-process event bus: publishers and subscribers only know the bus and the topic.
class Bus {
  readonly #subs = new Map<string, { id: number; name: string; fn: Handler }[]>();
  #nextId = 0;

  subscribe(topic: string, name: string, fn: Handler): number {
    const id = ++this.#nextId;
    this.#subs.set(topic, [...(this.#subs.get(topic) ?? []), { id, name, fn }]);
    return id;
  }

  unsubscribe(id: number): void {
    for (const [topic, list] of this.#subs) this.#subs.set(topic, list.filter((s) => s.id !== id));
  }

  // Calls every subscriber in order. One failing subscriber doesn't stop the others.
  publish(topic: string, e: OrderEvent): void {
    console.log(`publish ${topic} #${e.orderId}`);
    for (const s of this.#subs.get(topic) ?? []) {
      try {
        s.fn(e);
      } catch (err) {
        console.log(`  ${s.name.padEnd(9)} : failed: ${(err as Error).message}`);
      }
    }
  }
}

const bus = new Bus();

// The order service only publishes. It doesn't know who listens.
bus.subscribe("order.placed", "sms", (e) => {
  if (e.orderId === 503) throw new Error("gateway down");
  console.log(`  ${"sms".padEnd(9)} : sent "order #${e.orderId} confirmed" to ${e.customer}`);
});

const points = new Map<string, number>();
const loyalty =
  (sign: number): Handler =>
  (e) => {
    const p = sign * Math.floor(e.amount / 100); // 1 point per 100 taka
    points.set(e.customer, (points.get(e.customer) ?? 0) + p);
    const signed = p >= 0 ? `+${p}` : `${p}`;
    console.log(`  ${"loyalty".padEnd(9)} : ${e.customer} ${signed} points (total ${points.get(e.customer)})`);
  };
bus.subscribe("order.placed", "loyalty", loyalty(+1));
bus.subscribe("order.cancelled", "loyalty", loyalty(-1));

const analytics = bus.subscribe("order.placed", "analytics", (e) =>
  console.log(`  ${"analytics".padEnd(9)} : recorded #${e.orderId}`),
);

bus.publish("order.placed", { orderId: 501, customer: "Rahim", amount: 720 });
bus.publish("order.placed", { orderId: 502, customer: "Nadia", amount: 1500 });

bus.unsubscribe(analytics);
console.log("analytics unsubscribed");

bus.publish("order.placed", { orderId: 503, customer: "Rahim", amount: 300 });
bus.publish("order.cancelled", { orderId: 502, customer: "Nadia", amount: 1500 });

use std::cell::RefCell;
use std::collections::HashMap;
use std::rc::Rc;

#[derive(Clone)]
struct Event { order_id: u32, customer: String, amount: i32 } // amount in taka

type Handler = Box<dyn Fn(&Event) -> Result<(), String>>;

// An in-process event bus: publishers and subscribers only know the bus and the topic.
#[derive(Default)]
struct Bus { subs: HashMap<String, Vec<(u32, String, Handler)>>, next_id: u32 }

impl Bus {
    fn subscribe(&mut self, topic: &str, name: &str, f: Handler) -> u32 {
        self.next_id += 1;
        self.subs.entry(topic.into()).or_default().push((self.next_id, name.into(), f));
        self.next_id
    }
    fn unsubscribe(&mut self, id: u32) {
        for list in self.subs.values_mut() { list.retain(|(i, _, _)| *i != id); }
    }
    // Calls every subscriber in order. One failing subscriber doesn't stop the others.
    fn publish(&self, topic: &str, e: &Event) {
        println!("publish {topic} #{}", e.order_id);
        for (_, name, f) in self.subs.get(topic).into_iter().flatten() {
            if let Err(err) = f(e) { println!("  {name:<9} : failed: {err}"); }
        }
    }
}

fn main() {
    let mut bus = Bus::default();

    // The order service only publishes. It doesn't know who listens.
    bus.subscribe("order.placed", "sms", Box::new(|e| {
        if e.order_id == 503 { return Err("gateway down".into()); }
        println!("  {:<9} : sent \"order #{} confirmed\" to {}", "sms", e.order_id, e.customer);
        Ok(())
    }));

    // Two handlers share the points table, so it lives behind Rc<RefCell<...>>.
    let points = Rc::new(RefCell::new(HashMap::<String, i32>::new()));
    let loyalty = |sign: i32| -> Handler {
        let points = Rc::clone(&points);
        Box::new(move |e: &Event| {
            let p = sign * (e.amount / 100); // 1 point per 100 taka
            let mut pts = points.borrow_mut();
            let total = pts.entry(e.customer.clone()).or_insert(0);
            *total += p;
            println!("  {:<9} : {} {:+} points (total {})", "loyalty", e.customer, p, total);
            Ok(())
        })
    };
    bus.subscribe("order.placed", "loyalty", loyalty(1));
    bus.subscribe("order.cancelled", "loyalty", loyalty(-1));

    let analytics = bus.subscribe("order.placed", "analytics", Box::new(|e| {
        println!("  {:<9} : recorded #{}", "analytics", e.order_id);
        Ok(())
    }));

    let ev = |id, c: &str, a| Event { order_id: id, customer: c.into(), amount: a };
    bus.publish("order.placed", &ev(501, "Rahim", 720));
    bus.publish("order.placed", &ev(502, "Nadia", 1500));

    bus.unsubscribe(analytics);
    println!("analytics unsubscribed");

    bus.publish("order.placed", &ev(503, "Rahim", 300));
    bus.publish("order.cancelled", &ev(502, "Nadia", 1500));
}

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

public class Main {
    record Event(int orderId, String customer, int amount) {} // amount in taka

    record Subscription(int id, String name, Consumer<Event> fn) {}

    // An in-process event bus: publishers and subscribers only know the bus and the topic.
    static final class Bus {
        private final Map<String, List<Subscription>> subs = new HashMap<>();
        private int nextId = 0;

        int subscribe(String topic, String name, Consumer<Event> fn) {
            subs.computeIfAbsent(topic, t -> new ArrayList<>()).add(new Subscription(++nextId, name, fn));
            return nextId;
        }

        void unsubscribe(int id) {
            for (List<Subscription> list : subs.values()) list.removeIf(s -> s.id() == id);
        }

        // Calls every subscriber in order. One failing subscriber doesn't stop the others.
        void publish(String topic, Event e) {
            System.out.printf("publish %s #%d%n", topic, e.orderId());
            for (Subscription s : subs.getOrDefault(topic, List.of())) {
                try {
                    s.fn().accept(e);
                } catch (RuntimeException err) {
                    System.out.printf("  %-9s : failed: %s%n", s.name(), err.getMessage());
                }
            }
        }
    }

    public static void main(String[] args) {
        Bus bus = new Bus();

        // The order service only publishes. It doesn't know who listens.
        bus.subscribe("order.placed", "sms", e -> {
            if (e.orderId() == 503) throw new IllegalStateException("gateway down");
            System.out.printf("  %-9s : sent \"order #%d confirmed\" to %s%n", "sms", e.orderId(), e.customer());
        });

        Map<String, Integer> points = new HashMap<>();
        java.util.function.IntFunction<Consumer<Event>> loyalty = sign -> e -> {
            int p = sign * (e.amount() / 100); // 1 point per 100 taka
            int total = points.merge(e.customer(), p, Integer::sum);
            System.out.printf("  %-9s : %s %+d points (total %d)%n", "loyalty", e.customer(), p, total);
        };
        bus.subscribe("order.placed", "loyalty", loyalty.apply(1));
        bus.subscribe("order.cancelled", "loyalty", loyalty.apply(-1));

        int analytics = bus.subscribe("order.placed", "analytics",
            e -> System.out.printf("  %-9s : recorded #%d%n", "analytics", e.orderId()));

        bus.publish("order.placed", new Event(501, "Rahim", 720));
        bus.publish("order.placed", new Event(502, "Nadia", 1500));

        bus.unsubscribe(analytics);
        System.out.println("analytics unsubscribed");

        bus.publish("order.placed", new Event(503, "Rahim", 300));
        bus.publish("order.cancelled", new Event(502, "Nadia", 1500));
    }
}

#include <algorithm>
#include <cstdio>
#include <functional>
#include <map>
#include <stdexcept>
#include <string>
#include <vector>

struct Event {
    int orderId;
    std::string customer;
    int amount;  // taka
};

using Handler = std::function<void(const Event&)>;

// An in-process event bus: publishers and subscribers only know the bus and the topic.
class Bus {
public:
    int subscribe(const std::string& topic, const std::string& name, Handler fn) {
        subs_[topic].push_back({++nextId_, name, std::move(fn)});
        return nextId_;
    }
    void unsubscribe(int id) {
        for (auto& [topic, list] : subs_) std::erase_if(list, [&](const Sub& s) { return s.id == id; });
    }
    // Calls every subscriber in order. One failing subscriber doesn't stop the others.
    void publish(const std::string& topic, const Event& e) {
        std::printf("publish %s #%d\n", topic.c_str(), e.orderId);
        for (const auto& s : subs_[topic]) {
            try {
                s.fn(e);
            } catch (const std::exception& err) {
                std::printf("  %-9s : failed: %s\n", s.name.c_str(), err.what());
            }
        }
    }

private:
    struct Sub {
        int id;
        std::string name;
        Handler fn;
    };
    std::map<std::string, std::vector<Sub>> subs_;
    int nextId_ = 0;
};

int main() {
    Bus bus;

    // The order service only publishes. It doesn't know who listens.
    bus.subscribe("order.placed", "sms", [](const Event& e) {
        if (e.orderId == 503) throw std::runtime_error("gateway down");
        std::printf("  %-9s : sent \"order #%d confirmed\" to %s\n", "sms", e.orderId, e.customer.c_str());
    });

    std::map<std::string, int> points;
    auto loyalty = [&points](int sign) -> Handler {
        return [&points, sign](const Event& e) {
            int p = sign * (e.amount / 100);  // 1 point per 100 taka
            int total = (points[e.customer] += p);
            std::printf("  %-9s : %s %+d points (total %d)\n", "loyalty", e.customer.c_str(), p, total);
        };
    };
    bus.subscribe("order.placed", "loyalty", loyalty(1));
    bus.subscribe("order.cancelled", "loyalty", loyalty(-1));

    int analytics = bus.subscribe("order.placed", "analytics",
                                  [](const Event& e) { std::printf("  %-9s : recorded #%d\n", "analytics", e.orderId); });

    bus.publish("order.placed", {501, "Rahim", 720});
    bus.publish("order.placed", {502, "Nadia", 1500});

    bus.unsubscribe(analytics);
    std::printf("analytics unsubscribed\n");

    bus.publish("order.placed", {503, "Rahim", 300});
    bus.publish("order.cancelled", {502, "Nadia", 1500});
}