| 2 | (window.GOFRAMEWORK_FLOWS = window.GOFRAMEWORK_FLOWS || []).push({"id":"events","title":"Events with slow and failing subscribers","group":"concurrency","summary":"An order topic on an events.Bus with four subscribers: one called inside Publish, one slow with a small queue that drops the oldest, one that refuses when full, one that panics. Six orders arrive in a burst; then the program closes the bus, which finishes what is queued.","scenario":"crashproof.OnPanic(logPanic) // once, in main: panics are reported, not fatal\n\nbus := events.New()\norders := events.NewTopic[Order](\"orders\")\n\norders.Subscribe(bus, audit) // inside Publish\norders.Subscribe(bus, mail, events.Async(4, events.DropOldest)) // 3 s each\norders.Subscribe(bus, index, events.Async(2, events.Reject)) // 1 s each\norders.Subscribe(bus, invoice, events.Async(4, events.Block),\n\tevents.OnError(report)) // 0.5 s; panics on #4\n\nfor _, order := range burst {\n\terr := orders.Publish(ctx, bus, order)\n}\nbus.Close(ctx)","recorded":{"date":"2026-10-17","go":"go1.27.2","clock":"testing/synctest's fake clock: the times are the program's own, without waiting for them","modules":{"crashproof":"9f8a7e5","dsa-collections-queue-concurrent":"c112718","events":"09d672b","flow":"56cc385"}},"width":1100,"height":620,"nodes":[{"id":"publisher","kind":"outside","title":"Your code","x":20,"y":245,"w":210,"h":130},{"id":"bus","kind":"decision","title":"Bus: the orders topic","module":"events","x":300,"y":245,"w":220,"h":130},{"id":"audit","kind":"effect","title":"audit: Sync","module":"events","x":610,"y":10,"w":270,"h":105},{"id":"mailer","kind":"state","title":"mailer: Async(4, DropOldest)","module":"events","x":610,"y":130,"w":270,"h":150},{"id":"search","kind":"state","title":"search: Async(2, Reject)","module":"events","x":610,"y":295,"w":270,"h":150},{"id":"invoices","kind":"state","title":"invoices: Async(4, Block)","module":"events","x":610,"y":460,"w":270,"h":150},{"id":"onerror","kind":"outside","title":"Your handlers","x":915,"y":460,"w":175,"h":150}],"edges":[{"id":1,"from":"publisher","to":"bus","label":"Publish","bend":-12},{"id":2,"from":"bus","to":"audit","label":"call the handler, inside Publish"},{"id":3,"from":"bus","to":"mailer","label":"queue"},{"id":4,"from":"bus","to":"search","label":"queue"},{"id":5,"from":"bus","to":"invoices","label":"queue"},{"id":6,"from":"bus","to":"publisher","label":"an error back: ErrFull, ErrClosed","bend":-12},{"id":7,"from":"mailer","to":"mailer","label":"drop the oldest"},{"id":8,"from":"invoices","to":"onerror","label":"a panic: OnPanic, then OnError"},{"id":9,"from":"publisher","to":"bus","label":"Close(ctx)","bend":26}],"events":[{"chapter":0,"at":0,"node":"publisher","kind":"onpanic","text":"In main, your code sets crashproof.OnPanic(logPanic): a panic anywhere is logged instead of crashing the program.","state":{"bus":{"lines":[["published","0"],["queued now","0"],["dropped","0"]]},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","0"],["errors back","0"]]}}},{"chapter":0,"at":0,"edge":2,"node":"bus","kind":"subscribe","text":"audit subscribes: Sync, the default. Its handler runs inside Publish.","state":{"audit":{"lines":[["delivered","0"]]},"bus":{"lines":[["published","0"],["queued now","0"],["dropped","0"]]},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","0"],["errors back","0"]]}}},{"chapter":0,"at":0,"edge":3,"node":"mailer","kind":"subscribe","text":"mailer subscribes: Async(4, DropOldest). Sending a mail takes 3 s.","state":{"audit":{"lines":[["delivered","0"]]},"bus":{"lines":[["published","0"],["queued now","0"],["dropped","0"]]},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","0"],["errors back","0"]]}}},{"chapter":0,"at":0,"edge":4,"node":"search","kind":"subscribe","text":"search subscribes: Async(2, Reject). Indexing takes 1 s.","state":{"audit":{"lines":[["delivered","0"]]},"bus":{"lines":[["published","0"],["queued now","0"],["dropped","0"]]},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","0"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":0,"at":0,"edge":5,"node":"invoices","kind":"subscribe","text":"invoices subscribes: Async(4, Block), with an OnError. It takes 0.5 s, and panics on order #4.","state":{"audit":{"lines":[["delivered","0"]]},"bus":{"lines":[["published","0"],["queued now","0"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","0"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1,"edge":1,"node":"bus","kind":"publish","text":"Your code publishes order #1: orders.Publish(ctx, bus, order).","state":{"audit":{"lines":[["delivered","0"]]},"bus":{"lines":[["published","0"],["queued now","0"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1,"edge":2,"node":"audit","kind":"sync-call","text":"audit writes order #1 down inside Publish, on your goroutine; Publish waits for it.","state":{"audit":{"lines":[["delivered","0"]]},"bus":{"lines":[["published","1"],["queued now","0"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1,"edge":3,"node":"mailer","kind":"queued","text":"Order #1 goes into mailer's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","2"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":["#1"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1,"edge":4,"node":"search","kind":"queued","text":"Order #1 goes into search's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","1"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":["#1"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":["#1"]}}}},{"chapter":1,"at":1,"edge":5,"node":"invoices","kind":"queued","text":"Order #1 goes into invoices' queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","1"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":["#1"]}},"mailer":{"lines":[["delivered","0"],["dropped","0"]],"cells":{"label":"queue","size":4,"items":["#1"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":["#1"]}}}},{"chapter":1,"at":1,"node":"mailer","kind":"handling","text":"mailer's worker takes #1 from its queue and handles it (3 s).","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","1"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":["#1"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"lines":[["delivered","0"],["rejected","0"]],"cells":{"label":"queue","size":2,"items":["#1"]}}}},{"chapter":1,"at":1,"node":"search","kind":"handling","text":"search's worker takes #1 from its queue and handles it (1 s).","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","1"],["dropped","0"]]},"invoices":{"lines":[["delivered","0"],["panicked","0"]],"cells":{"label":"queue","size":4,"items":["#1"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1,"node":"invoices","kind":"handling","text":"invoices' worker takes #1 from its queue and handles it (0.5 s).","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","0"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","1"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1.2,"edge":1,"node":"bus","kind":"publish","text":"Your code publishes order #2: orders.Publish(ctx, bus, order).","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","1"],["queued now","0"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","2"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1.2,"edge":2,"node":"audit","kind":"sync-call","text":"audit writes order #2 down inside Publish, on your goroutine; Publish waits for it.","state":{"audit":{"lines":[["delivered","1"]]},"bus":{"lines":[["published","2"],["queued now","0"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","2"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1.2,"edge":3,"node":"mailer","kind":"queued","text":"Order #2 goes into mailer's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","2"]]},"bus":{"lines":[["published","2"],["queued now","3"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","2"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":1,"at":1.2,"edge":4,"node":"search","kind":"queued","text":"Order #2 goes into search's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","2"]]},"bus":{"lines":[["published","2"],["queued now","3"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","2"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2"]}}}},{"chapter":1,"at":1.2,"edge":5,"node":"invoices","kind":"queued","text":"Order #2 goes into invoices' queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","2"]]},"bus":{"lines":[["published","2"],["queued now","3"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","2"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2"]}}}},{"chapter":1,"at":1.4,"edge":1,"node":"bus","kind":"publish","text":"Your code publishes order #3: orders.Publish(ctx, bus, order).","state":{"audit":{"lines":[["delivered","2"]]},"bus":{"lines":[["published","2"],["queued now","3"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","3"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2"]}}}},{"chapter":1,"at":1.4,"edge":2,"node":"audit","kind":"sync-call","text":"audit writes order #3 down inside Publish, on your goroutine; Publish waits for it.","state":{"audit":{"lines":[["delivered","2"]]},"bus":{"lines":[["published","3"],["queued now","3"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","3"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2"]}}}},{"chapter":1,"at":1.4,"edge":3,"node":"mailer","kind":"queued","text":"Order #3 goes into mailer's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","3"]]},"bus":{"lines":[["published","3"],["queued now","6"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","3"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2"]}}}},{"chapter":1,"at":1.4,"edge":4,"node":"search","kind":"queued","text":"Order #3 goes into search's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","3"]]},"bus":{"lines":[["published","3"],["queued now","6"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","3"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.4,"edge":5,"node":"invoices","kind":"queued","text":"Order #3 goes into invoices' queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","3"]]},"bus":{"lines":[["published","3"],["queued now","6"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","0"],["panicked","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","3"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.5,"node":"invoices","kind":"handling","text":"invoices' worker takes #2 from its queue and handles it (0.5 s).","state":{"audit":{"lines":[["delivered","3"]]},"bus":{"lines":[["published","3"],["queued now","5"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","3"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.6,"edge":1,"node":"bus","kind":"publish","text":"Your code publishes order #4: orders.Publish(ctx, bus, order).","state":{"audit":{"lines":[["delivered","3"]]},"bus":{"lines":[["published","3"],["queued now","5"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","4"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.6,"edge":2,"node":"audit","kind":"sync-call","text":"audit writes order #4 down inside Publish, on your goroutine; Publish waits for it.","state":{"audit":{"lines":[["delivered","3"]]},"bus":{"lines":[["published","4"],["queued now","5"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","4"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","0"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.6,"edge":3,"node":"mailer","kind":"queued","text":"Order #4 goes into mailer's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","4"]]},"bus":{"lines":[["published","4"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","4"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","1"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.6,"edge":4,"node":"search","kind":"rejected","text":"search's queue is full (2 of 2): Reject refuses order #4, and Publish will return ErrFull.","state":{"audit":{"lines":[["delivered","4"]]},"bus":{"lines":[["published","4"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","4"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","1"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.6,"edge":5,"node":"invoices","kind":"queued","text":"Order #4 goes into invoices' queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","4"]]},"bus":{"lines":[["published","4"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","4"],["errors back","0"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","1"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.6,"edge":6,"node":"publisher","kind":"rejected","text":"Publish returns an error for order #4, \"concurrent: the queue is full\": your code knows search did not take it.","state":{"audit":{"lines":[["delivered","4"]]},"bus":{"lines":[["published","4"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","4"],["errors back","1"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","1"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.8,"edge":1,"node":"bus","kind":"publish","text":"Your code publishes order #5: orders.Publish(ctx, bus, order).","state":{"audit":{"lines":[["delivered","4"]]},"bus":{"lines":[["published","4"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","1"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","1"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.8,"edge":2,"node":"audit","kind":"sync-call","text":"audit writes order #5 down inside Publish, on your goroutine; Publish waits for it.","state":{"audit":{"lines":[["delivered","4"]]},"bus":{"lines":[["published","5"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","1"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","1"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.8,"edge":3,"node":"mailer","kind":"queued","text":"Order #5 goes into mailer's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","9"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","1"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","2"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.8,"edge":4,"node":"search","kind":"rejected","text":"search's queue is full (2 of 2): Reject refuses order #5, and Publish will return ErrFull.","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","9"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","1"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","2"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.8,"edge":5,"node":"invoices","kind":"queued","text":"Order #5 goes into invoices' queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","9"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","1"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","2"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":1.8,"edge":6,"node":"publisher","kind":"rejected","text":"Publish returns an error for order #5, \"concurrent: the queue is full\": your code knows search did not take it.","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","9"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","0"],["rejected","2"],["handling","#1"]],"cells":{"label":"queue","size":2,"items":["#2","#3"]}}}},{"chapter":1,"at":2,"node":"search","kind":"handling","text":"search's worker takes #2 from its queue and handles it (1 s).","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","8"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","1"],["panicked","0"],["handling","#2"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3"]}}}},{"chapter":1,"at":2,"node":"invoices","kind":"handling","text":"invoices' worker takes #3 from its queue and handles it (0.5 s).","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","2"],["panicked","0"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","5"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3"]}}}},{"chapter":1,"at":2.1,"edge":1,"node":"bus","kind":"publish","text":"Your code publishes order #6: orders.Publish(ctx, bus, order).","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","5"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","2"],["panicked","0"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3"]}}}},{"chapter":1,"at":2.1,"edge":2,"node":"audit","kind":"sync-call","text":"audit writes order #6 down inside Publish, on your goroutine; Publish waits for it.","state":{"audit":{"lines":[["delivered","5"]]},"bus":{"lines":[["published","6"],["queued now","7"],["dropped","0"]]},"invoices":{"status":"busy","lines":[["delivered","2"],["panicked","0"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","0"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#2","#3","#4","#5"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3"]}}}},{"chapter":1,"at":2.1,"edge":7,"node":"mailer","kind":"dropped","text":"mailer's queue is full: order #2, the oldest, is dropped to make room for #6.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","9"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","2"],["panicked","0"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3"]}}}},{"chapter":1,"at":2.1,"edge":4,"node":"search","kind":"queued","text":"Order #6 goes into search's queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","9"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","2"],["panicked","0"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":1,"at":2.1,"edge":5,"node":"invoices","kind":"queued","text":"Order #6 goes into invoices' queue; Publish does not wait for it.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","9"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","2"],["panicked","0"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5","#6"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":1,"at":2.5,"node":"invoices","kind":"handling","text":"invoices' worker takes #4 from its queue and handles it (0.5 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","8"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","3"],["panicked","0"],["handling","#4"]],"cells":{"label":"queue","size":4,"items":["#5","#6"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":2,"at":2.5,"edge":8,"node":"onerror","kind":"panic-logged","text":"crashproof's handler gets the panic and its stack: \"invoice #4: no tax rate for its country\", raised in the invoices handler. Your code logs it.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","8"],["dropped","1"]]},"invoices":{"lines":[["delivered","3"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":["#5","#6"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","0"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":2,"at":2.5,"edge":8,"node":"onerror","kind":"panic","text":"invoices' handler panicked (\"invoice #4: no tax rate for its country\"). The bus recovered it, reported it here with its stack, and the worker goes on.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","8"],["dropped","1"]]},"invoices":{"lines":[["delivered","3"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":["#5","#6"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":2,"at":2.5,"node":"invoices","kind":"handling","text":"invoices' worker takes #5 from its queue and handles it (0.5 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","7"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","3"],["panicked","1"],["handling","#5"]],"cells":{"label":"queue","size":4,"items":["#6"]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":2,"at":3,"node":"invoices","kind":"handling","text":"invoices' worker takes #6 from its queue and handles it (0.5 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","6"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","4"],["panicked","1"],["handling","#6"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","1"],["rejected","2"],["handling","#2"]],"cells":{"label":"queue","size":2,"items":["#3","#6"]}}}},{"chapter":2,"at":3,"node":"search","kind":"handling","text":"search's worker takes #3 from its queue and handles it (1 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","5"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","4"],["panicked","1"],["handling","#6"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","2"],["rejected","2"],["handling","#3"]],"cells":{"label":"queue","size":2,"items":["#6"]}}}},{"chapter":3,"at":3.2,"edge":9,"node":"bus","kind":"close","text":"Your code closes the bus: bus.Close(ctx). No new events; the queues are finished first.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","5"],["dropped","1"]]},"invoices":{"status":"busy","lines":[["delivered","4"],["panicked","1"],["handling","#6"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","0"],["dropped","1"],["handling","#1"]],"cells":{"label":"queue","size":4,"items":["#3","#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","2"],["rejected","2"],["handling","#3"]],"cells":{"label":"queue","size":2,"items":["#6"]}}}},{"chapter":3,"at":4,"node":"mailer","kind":"handling","text":"mailer's worker takes #3 from its queue and handles it (3 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","4"],["dropped","1"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","1"],["dropped","1"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","2"],["rejected","2"],["handling","#3"]],"cells":{"label":"queue","size":2,"items":["#6"]}}}},{"chapter":3,"at":4,"node":"search","kind":"handling","text":"search's worker takes #6 from its queue and handles it (1 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","3"],["dropped","1"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","1"],["dropped","1"],["handling","#3"]],"cells":{"label":"queue","size":4,"items":["#4","#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"status":"busy","lines":[["delivered","3"],["rejected","2"],["handling","#6"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":3,"at":7,"node":"mailer","kind":"handling","text":"mailer's worker takes #4 from its queue and handles it (3 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","2"],["dropped","1"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","2"],["dropped","1"],["handling","#4"]],"cells":{"label":"queue","size":4,"items":["#5","#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"lines":[["delivered","4"],["rejected","2"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":3,"at":10,"node":"mailer","kind":"handling","text":"mailer's worker takes #5 from its queue and handles it (3 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","1"],["dropped","1"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","3"],["dropped","1"],["handling","#5"]],"cells":{"label":"queue","size":4,"items":["#6"]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"lines":[["delivered","4"],["rejected","2"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":3,"at":13,"node":"mailer","kind":"handling","text":"mailer's worker takes #6 from its queue and handles it (3 s).","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"lines":[["published","6"],["queued now","0"],["dropped","1"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"status":"busy","lines":[["delivered","4"],["dropped","1"],["handling","#6"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"lines":[["delivered","4"],["rejected","2"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":3,"at":16,"node":"bus","kind":"closed","text":"Close returned \u003cnil\u003e after 12.8 s: every queued order was handled. Delivered 20, panicked 1, dropped 1, rejected 2.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"status":"off","lines":[["published","6"],["queued now","0"],["dropped","1"],["state","closed"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","5"],["dropped","1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","6"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"lines":[["delivered","4"],["rejected","2"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":3,"at":16,"edge":1,"node":"bus","kind":"publish-closed","text":"A late publish, order #7, after Close.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"status":"off","lines":[["published","6"],["queued now","0"],["dropped","1"],["state","closed"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","5"],["dropped","1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","7"],["errors back","2"],["last","concurrent: the queue is full"]]},"search":{"lines":[["delivered","4"],["rejected","2"]],"cells":{"label":"queue","size":2,"items":[]}}}},{"chapter":3,"at":16,"edge":6,"node":"publisher","kind":"publish-closed","text":"Publish returns \"events: the bus is closed\": the bus is gone, and your code is told so.","state":{"audit":{"lines":[["delivered","6"]]},"bus":{"status":"off","lines":[["published","6"],["queued now","0"],["dropped","1"],["state","closed"]]},"invoices":{"lines":[["delivered","5"],["panicked","1"]],"cells":{"label":"queue","size":4,"items":[]}},"mailer":{"lines":[["delivered","5"],["dropped","1"]],"cells":{"label":"queue","size":4,"items":[]}},"onerror":{"lines":[["reports","1"]]},"publisher":{"lines":[["orders published","7"],["errors back","3"],["last","events: the bus is closed"]]},"search":{"lines":[["delivered","4"],["rejected","2"]],"cells":{"label":"queue","size":2,"items":[]}}}}],"chapters":["Four subscribers, each choosing what happens when its queue is full","A burst of six orders: queues fill, the oldest is dropped, one refuses","A handler panics: only that subscriber is hurt","Your code closes the bus: every queued order is handled first"],"answers":{"close":{"title":"Close","why":"Exiting with events still queued loses them. Close stops new events at once, lets every worker finish what is queued, and waits, up to the context's deadline, until no handler runs.","how":"Close marks the bus closed, closes every queue (workers drain them, then end), waits for the workers or the context, then for running handlers.","code":[{"module":"events","file":"events.go","symbol":"Bus.Close","line":259,"source":"// Close stops publishing (ErrClosed from then on), lets the asynchronous\n// subscriptions handle what they have queued, closes every subscription and\n// waits for the handler calls still running: once it returns nil, no\n// handler is running and none will start. If ctx ends first, the handlers'\n// contexts end too, and Close says how many events were left. Calling it\n// from a handler waits for that handler, so until ctx ends.\nfunc (thisRef *Bus) Close(ctx context.Context) error {\n\tthisRef.mutex.Lock()\n\tif thisRef.closed.Load() {\n\t\tthisRef.mutex.Unlock()\n\t\treturn nil\n\t}\n\tthisRef.closed.Store(true)\n\tsubscriptions := slices.Collect(maps.Keys(thisRef.subscriptions))\n\tthisRef.mutex.Unlock()\n\n\tfor _, each := range subscriptions {\n\t\teach.typed.closeQueue()\n\t}\n\tclose(thisRef.closing)\n\tvar err error\n\tselect {\n\tcase \u003c-thisRef.ended:\n\tcase \u003c-ctx.Done():\n\t\tleft := 0\n\t\tfor _, each := range subscriptions {\n\t\t\tleft += each.typed.queued()\n\t\t}\n\t\terr = fmt.Errorf(\"events: %d events not handled: %w\", left, context.Cause(ctx))\n\t}\n\t// …\n}"}]},"closed":{"title":"Closed","why":"Every published event is accounted for: delivered, failed, panicked, dropped or rejected; Stats() adds up.","how":"Bus.Stats sums every subscription's counts.","code":[{"module":"events","file":"events.go","symbol":"Bus.Stats","line":329,"source":"// Stats returns the bus's counts: Published counts Publish calls, the rest\n// add up every subscription's, closed ones included.\nfunc (thisRef *Bus) Stats() Stats {\n\tthisRef.mutex.Lock()\n\ttotal := thisRef.retired\n\tfor each := range thisRef.subscriptions {\n\t\ttotal.add(each.Stats())\n\t}\n\tthisRef.mutex.Unlock()\n\ttotal.Published = thisRef.published.Load()\n\treturn total\n}"}]},"dropped":{"title":"Full: the oldest is dropped","why":"For a mailer, the newest orders matter more than old ones it cannot catch up with, and the publisher must never wait for it. DropOldest keeps the queue's newest events and counts what it drops, so the loss shows in Stats().","how":"Push finds the queue full; with DropOldest it pops the oldest value, counts it as dropped, and pushes the new one.","code":[{"module":"dsa-collections-queue-concurrent","file":"queue.go","symbol":"Queue.Push","line":253,"source":"// Push queues value; when the queue is full, it does what the policy says:\n// Block waits for room and returns ctx's error if ctx ends first, Reject\n// returns ErrFull, the drops return nil. ErrClosed once the queue is\n// closed.\nfunc (thisRef *Queue[T]) Push(ctx context.Context, value T) error {\n\tfor {\n\t\tpushed, closed := thisRef.tryPush(value)\n\t\tif pushed {\n\t\t\tif thisRef.notEmpty.count.Load() != 0 {\n\t\t\t\tthisRef.notEmpty.wake(1)\n\t\t\t}\n\t\t\treturn nil\n\t\t}\n\t\tif closed {\n\t\t\treturn ErrClosed\n\t\t}\n\t\tif !thisRef.full() {\n\t\t\t// The slot is still being popped: the queue is not full, so\n\t\t\t// wait for that pop rather than drop or refuse.\n\t\t\truntime.Gosched()\n\t\t\tcontinue\n\t\t}\n\t\tswitch thisRef.whenFull {\n\t\tcase Reject:\n\t\t\treturn ErrFull\n\t\tcase DropNewest:\n\t\t\tthisRef.drop(value)\n\t\t\treturn nil\n\t\tcase DropOldest:\n\t\t\tif oldest, popped := thisRef.tryPop(); popped {\n\t// …\n}"}]},"handling":{"title":"A worker handles an event","why":"One worker per subscription by default keeps its events in order. The worker takes everything queued at once, so a burst costs one wake-up, not one per event.","how":"work pops a batch with PopInto and calls the handler for each, in order; a handler's error or panic goes to OnError.","code":[{"module":"events","file":"subscription.go","symbol":"subscriber.work","line":217,"source":"// work handles queued events until the queue is closed and empty, or ctx\n// ends; the last worker to end retires the subscription.\nfunc (thisRef *subscriber[T]) work(ctx context.Context) error {\n\tdefer func() {\n\t\tif thisRef.running.Add(-1) == 0 {\n\t\t\tthisRef.bus.retire(thisRef.Subscription)\n\t\t}\n\t}()\n\tbatch := make([]T, thisRef.batch)\n\tvar none T\n\tfor {\n\t\t// Not ctx: every way the workers end closes the queue first, and a\n\t\t// hundred workers waiting on one context's channel contend for its\n\t\t// lock (measured: three times slower at a hundred subscribers).\n\t\ttaken, err := thisRef.queue.PopInto(context.Background(), batch)\n\t\tif err != nil {\n\t\t\treturn nil\n\t\t}\n\t\tif taken \u003e 1 {\n\t\t\tthisRef.pending.Add(int64(taken - 1))\n\t\t}\n\t\tfor index := range taken {\n\t\t\tif index \u003e 0 {\n\t\t\t\tthisRef.pending.Add(-1)\n\t\t\t}\n\t\t\tcalled, err := thisRef.call(ctx, batch[index])\n\t\t\tswitch {\n\t\t\tcase !called:\n\t\t\t\tthisRef.discarded.Add(1)\n\t\t\tcase err != nil \u0026\u0026 thisRef.onError != nil:\n\t// …\n}"}]},"onpanic":{"title":"Panics, reported once","why":"goframework never swallows a panic on its own: without a handler, a recovered panic is raised again and the program crashes, as an unhandled panic would. The application decides, once in main, that panics are reported instead, typically to its log.","how":"crashproof.OnPanic sets the one handler that every recovered panic goes to, with the stack where it was raised.","code":[{"module":"crashproof","file":"go.go","symbol":"OnPanic","line":47,"source":"// OnPanic sets the function that receives every panic caught by Go and Run,\n// in every goroutine, from now on. It is meant to be called once, by the\n// application: a library that calls it replaces the application's handler.\n// nil removes the handler, and panics crash the program again.\nfunc OnPanic(handle func(Panic)) {\n\tif handle == nil {\n\t\thandler.Store(nil)\n\t\treturn\n\t}\n\thandler.Store(\u0026handle)\n}"},{"module":"crashproof","file":"go.go","symbol":"Recovered","line":87,"source":"// Recovered is for code that recovers a panic itself, to go on and keep its\n// value: called from the deferred function, with what recover returned, it\n// returns the panic with the stack where it was raised and sends it to the\n// OnPanic handler. Without a handler it panics again with the value, so the\n// program crashes as an unhandled panic does.\n//\n//\tdefer func() {\n//\t\tif value := recover(); value != nil {\n//\t\t\terr = fmt.Errorf(\"the task failed: %v\", crashproof.Recovered(value))\n//\t\t}\n//\t}()\nfunc Recovered(value any) Panic {\n\trecovered := Panic{Value: value, Frames: frames()}\n\thandle := handler.Load()\n\tif handle == nil {\n\t\tpanic(value)\n\t}\n\treport(*handle, recovered)\n\treturn recovered\n}"}]},"panic":{"title":"A handler panics","why":"A panic in any goroutine ends a Go program. A subscriber's bug must hurt only that subscriber: the bus recovers the panic, reports it with its stack, and the worker goes on with the next event.","how":"call recovers the panic into a *PanicError (through crashproof, which keeps its stack); work passes it to OnError.","code":[{"module":"events","file":"subscription.go","symbol":"subscriber.call","line":196,"source":"// call runs the handler with value unless the subscription is closed; a\n// panic becomes a *PanicError, after crashproof's handler saw it.\nfunc (thisRef *subscriber[T]) call(ctx context.Context, value T) (called bool, err error) {\n\tif !thisRef.enter() {\n\t\treturn false, nil\n\t}\n\tdefer thisRef.leave()\n\tdefer func() {\n\t\tif recovered := recover(); recovered != nil {\n\t\t\tthisRef.panicked.Add(1)\n\t\t\tcalled, err = true, \u0026PanicError{Panic: crashproof.Recovered(recovered)}\n\t\t}\n\t}()\n\tif err = thisRef.handler(ctx, value); err != nil {\n\t\tthisRef.failed.Add(1)\n\t\treturn true, err\n\t}\n\tthisRef.delivered.Add(1)\n\treturn true, nil\n}"},{"module":"events","file":"events.go","symbol":"type PanicError","line":48,"source":"// PanicError is the error of a handler that panicked: the value and the\n// stack where it was raised. The panic was sent to crashproof's handler\n// first; with none set, the program crashes, as everywhere in goframework.\ntype PanicError struct {\n\tcrashproof.Panic\n}"}]},"panic-logged":{"title":"The panic, logged","why":"A panic is a bug; its stack is what fixes it. The handler gets the stack where the panic was raised, not where it was recovered.","how":"events' call recovers the panic and passes it to crashproof.Recovered, which keeps the stack and calls the OnPanic handler.","code":[{"module":"crashproof","file":"go.go","symbol":"Recovered","line":87,"source":"// Recovered is for code that recovers a panic itself, to go on and keep its\n// value: called from the deferred function, with what recover returned, it\n// returns the panic with the stack where it was raised and sends it to the\n// OnPanic handler. Without a handler it panics again with the value, so the\n// program crashes as an unhandled panic does.\n//\n//\tdefer func() {\n//\t\tif value := recover(); value != nil {\n//\t\t\terr = fmt.Errorf(\"the task failed: %v\", crashproof.Recovered(value))\n//\t\t}\n//\t}()\nfunc Recovered(value any) Panic {\n\trecovered := Panic{Value: value, Frames: frames()}\n\thandle := handler.Load()\n\tif handle == nil {\n\t\tpanic(value)\n\t}\n\treport(*handle, recovered)\n\treturn recovered\n}"}]},"publish":{"title":"Publishing","why":"Publish is typed: a topic carries one Go type, checked by the compiler, not interface{} and a string name. It offers the event to every subscription in the order they subscribed, and returns their errors joined, so a refusal is never silent.","how":"Topic.Publish reads the topic's routes without a lock and offers the value to each subscriber.","code":[{"module":"events","file":"events.go","symbol":"Topic.Publish","line":114,"source":"// Publish hands value to every subscription of the topic on bus: it calls\n// the synchronous ones' handlers, in the order they subscribed, and queues\n// it for the asynchronous ones. It returns their errors, joined: the\n// synchronous handlers' own, ErrFull from a full Reject queue, ctx's error\n// from a full Block queue that ctx ended the wait for. ErrClosed once the\n// bus is closing.\nfunc (thisRef Topic[T]) Publish(ctx context.Context, bus *Bus, value T) error {\n\tthisRef.check()\n\tif bus.closed.Load() {\n\t\treturn ErrClosed\n\t}\n\tbus.published.Add(1)\n\troutes := *bus.routes.Load()\n\tif thisRef.id \u003e= len(routes) {\n\t\treturn nil\n\t}\n\tsubscribers, _ := routes[thisRef.id].([]*subscriber[T])\n\tvar errs []error\n\tfor _, each := range subscribers {\n\t\tif err := each.offer(ctx, value); err != nil {\n\t\t\terrs = append(errs, err)\n\t\t}\n\t}\n\tswitch len(errs) {\n\tcase 0:\n\t\treturn nil\n\tcase 1:\n\t\treturn errs[0]\n\t}\n\treturn errors.Join(errs...)\n}"},{"module":"events","file":"subscription.go","symbol":"subscriber.offer","line":173,"source":"// offer hands value to the subscription: calls the handler, or queues it.\n// It is counted as offered unless the subscription closed meanwhile, as if\n// it had been gone already; so once all is done, what was offered was\n// delivered, failed, panicked, dropped or rejected.\nfunc (thisRef *subscriber[T]) offer(ctx context.Context, value T) error {\n\tif thisRef.queue == nil {\n\t\t_, err := thisRef.call(ctx, value)\n\t\treturn err\n\t}\n\terr := thisRef.queue.Push(ctx, value)\n\tif err == nil {\n\t\tthisRef.offered.Add(1)\n\t\treturn nil\n\t}\n\tif errors.Is(err, concurrent.ErrClosed) {\n\t\tif thisRef.Closed() {\n\t\t\treturn nil\n\t\t}\n\t\terr = ErrClosed // the bus is closing\n\t}\n\tthisRef.offered.Add(1)\n\tthisRef.rejected.Add(1)\n\treturn err\n}"}]},"publish-closed":{"title":"Publishing after Close","why":"A late publisher learns the bus is gone instead of its event vanishing.","how":"Publish checks the bus's closed flag first and returns ErrClosed.","code":[{"module":"events","file":"events.go","symbol":"Topic.Publish","line":114,"source":"// Publish hands value to every subscription of the topic on bus: it calls\n// the synchronous ones' handlers, in the order they subscribed, and queues\n// it for the asynchronous ones. It returns their errors, joined: the\n// synchronous handlers' own, ErrFull from a full Reject queue, ctx's error\n// from a full Block queue that ctx ended the wait for. ErrClosed once the\n// bus is closing.\nfunc (thisRef Topic[T]) Publish(ctx context.Context, bus *Bus, value T) error {\n\tthisRef.check()\n\tif bus.closed.Load() {\n\t\treturn ErrClosed\n\t}\n\tbus.published.Add(1)\n\troutes := *bus.routes.Load()\n\tif thisRef.id \u003e= len(routes) {\n\t\treturn nil\n\t}\n\tsubscribers, _ := routes[thisRef.id].([]*subscriber[T])\n\tvar errs []error\n\tfor _, each := range subscribers {\n\t\tif err := each.offer(ctx, value); err != nil {\n\t\t\terrs = append(errs, err)\n\t\t}\n\t}\n\tswitch len(errs) {\n\tcase 0:\n\t\treturn nil\n\tcase 1:\n\t\treturn errs[0]\n\t}\n\treturn errors.Join(errs...)\n}"}]},"queued":{"title":"Queued","why":"A slow subscriber must not slow the publisher or the other subscribers. Each Async subscription has its own bounded queue: the publisher only pays for putting the event in it.","how":"offer pushes onto the subscription's queue, a lock-free bounded queue (Vyukov's algorithm); its worker is woken if it waits.","code":[{"module":"events","file":"subscription.go","symbol":"subscriber.offer","line":173,"source":"// offer hands value to the subscription: calls the handler, or queues it.\n// It is counted as offered unless the subscription closed meanwhile, as if\n// it had been gone already; so once all is done, what was offered was\n// delivered, failed, panicked, dropped or rejected.\nfunc (thisRef *subscriber[T]) offer(ctx context.Context, value T) error {\n\tif thisRef.queue == nil {\n\t\t_, err := thisRef.call(ctx, value)\n\t\treturn err\n\t}\n\terr := thisRef.queue.Push(ctx, value)\n\tif err == nil {\n\t\tthisRef.offered.Add(1)\n\t\treturn nil\n\t}\n\tif errors.Is(err, concurrent.ErrClosed) {\n\t\tif thisRef.Closed() {\n\t\t\treturn nil\n\t\t}\n\t\terr = ErrClosed // the bus is closing\n\t}\n\tthisRef.offered.Add(1)\n\tthisRef.rejected.Add(1)\n\treturn err\n}"},{"module":"dsa-collections-queue-concurrent","file":"queue.go","symbol":"Queue.Push","line":253,"source":"// Push queues value; when the queue is full, it does what the policy says:\n// Block waits for room and returns ctx's error if ctx ends first, Reject\n// returns ErrFull, the drops return nil. ErrClosed once the queue is\n// closed.\nfunc (thisRef *Queue[T]) Push(ctx context.Context, value T) error {\n\tfor {\n\t\tpushed, closed := thisRef.tryPush(value)\n\t\tif pushed {\n\t\t\tif thisRef.notEmpty.count.Load() != 0 {\n\t\t\t\tthisRef.notEmpty.wake(1)\n\t\t\t}\n\t\t\treturn nil\n\t\t}\n\t\tif closed {\n\t\t\treturn ErrClosed\n\t\t}\n\t\tif !thisRef.full() {\n\t\t\t// The slot is still being popped: the queue is not full, so\n\t\t\t// wait for that pop rather than drop or refuse.\n\t\t\truntime.Gosched()\n\t\t\tcontinue\n\t\t}\n\t\tswitch thisRef.whenFull {\n\t\tcase Reject:\n\t\t\treturn ErrFull\n\t\tcase DropNewest:\n\t\t\tthisRef.drop(value)\n\t\t\treturn nil\n\t\tcase DropOldest:\n\t\t\tif oldest, popped := thisRef.tryPop(); popped {\n\t// …\n}"}]},"rejected":{"title":"Full: refused","why":"Some subscribers would rather have the publisher know: Reject makes Publish return ErrFull, and the publisher decides (retry later, log, slow down). Nothing is lost silently.","how":"Push returns ErrFull; offer counts it as rejected and Publish returns it, joined with any other subscriber's error.","code":[{"module":"events","file":"subscription.go","symbol":"subscriber.offer","line":173,"source":"// offer hands value to the subscription: calls the handler, or queues it.\n// It is counted as offered unless the subscription closed meanwhile, as if\n// it had been gone already; so once all is done, what was offered was\n// delivered, failed, panicked, dropped or rejected.\nfunc (thisRef *subscriber[T]) offer(ctx context.Context, value T) error {\n\tif thisRef.queue == nil {\n\t\t_, err := thisRef.call(ctx, value)\n\t\treturn err\n\t}\n\terr := thisRef.queue.Push(ctx, value)\n\tif err == nil {\n\t\tthisRef.offered.Add(1)\n\t\treturn nil\n\t}\n\tif errors.Is(err, concurrent.ErrClosed) {\n\t\tif thisRef.Closed() {\n\t\t\treturn nil\n\t\t}\n\t\terr = ErrClosed // the bus is closing\n\t}\n\tthisRef.offered.Add(1)\n\tthisRef.rejected.Add(1)\n\treturn err\n}"}]},"subscribe":{"title":"Subscribing","why":"Each subscriber says how it wants events: inside Publish, or queued for its own goroutine, and what to do when its queue is full. That choice belongs to the subscriber, because only it knows whether it can lose an event, slow the publisher, or refuse.","how":"Subscribe makes a subscriber with its handler and options; an Async one gets a bounded queue (dsa-collections-queue-concurrent) and its workers.","code":[{"module":"events","file":"events.go","symbol":"Topic.Subscribe","line":143,"source":"// Subscribe calls handler with each event published on the topic on bus\n// from now on, as the options say: synchronously by default, inside\n// Publish. Subscribing to a closed bus returns a closed subscription.\nfunc (thisRef Topic[T]) Subscribe(bus *Bus, handler func(ctx context.Context, value T) error, options ...Option) *Subscription {\n\tthisRef.check()\n\tif handler == nil {\n\t\tpanic(\"events: a nil handler\")\n\t}\n\tset := settings{workers: 1}\n\tfor _, option := range options {\n\t\toption(\u0026set)\n\t}\n\tif set.workers \u003c 1 {\n\t\tpanic(fmt.Sprintf(\"events: Workers(%d)\", set.workers))\n\t}\n\tif set.workers \u003e 1 \u0026\u0026 !set.async {\n\t\tpanic(\"events: Workers needs Async\")\n\t}\n\tif set.async \u0026\u0026 set.size \u003c 1 {\n\t\tpanic(fmt.Sprintf(\"events: Async(%d, ...)\", set.size))\n\t}\n\tsubscription := \u0026Subscription{bus: bus, id: thisRef.id, topic: thisRef.name, name: set.name, onError: set.onError, synchronous: !set.async}\n\ttyped := \u0026subscriber[T]{Subscription: subscription, handler: handler}\n\tsubscription.typed = typed\n\tif set.async {\n\t\ttyped.queue = concurrent.New[T](set.size, set.whenFull)\n\t\ttyped.batch = 1\n\t\tif set.workers == 1 {\n\t\t\ttyped.batch = 32 // one worker keeps the order: take what is there\n\t\t}\n\t// …\n}"}]},"sync-call":{"title":"A synchronous subscriber","why":"Some work must happen before Publish returns, such as an audit record written before the order is confirmed. A synchronous subscriber runs on the publisher's goroutine, with its context, and its error comes back from Publish.","how":"offer calls the handler directly when the subscription has no queue; call turns a panic into a *PanicError.","code":[{"module":"events","file":"subscription.go","symbol":"subscriber.call","line":196,"source":"// call runs the handler with value unless the subscription is closed; a\n// panic becomes a *PanicError, after crashproof's handler saw it.\nfunc (thisRef *subscriber[T]) call(ctx context.Context, value T) (called bool, err error) {\n\tif !thisRef.enter() {\n\t\treturn false, nil\n\t}\n\tdefer thisRef.leave()\n\tdefer func() {\n\t\tif recovered := recover(); recovered != nil {\n\t\t\tthisRef.panicked.Add(1)\n\t\t\tcalled, err = true, \u0026PanicError{Panic: crashproof.Recovered(recovered)}\n\t\t}\n\t}()\n\tif err = thisRef.handler(ctx, value); err != nil {\n\t\tthisRef.failed.Add(1)\n\t\treturn true, err\n\t}\n\tthisRef.delivered.Add(1)\n\treturn true, nil\n}"}]}},"questions":[{"q":"Why not Go channels?","a":"A channel per subscriber gives no policy when it is full (it blocks), no counts, no panic recovery, and needs a goroutine and a select per subscriber written by hand. events does that once: bounded queues with a policy, Stats(), recovered panics, Close that drains."},{"q":"Does one slow subscriber slow the others?","a":"No. Each Async subscription has its own queue and workers; the publisher only puts the event in each queue. Only a Sync subscriber, or a Block queue that is full, makes Publish wait."},{"q":"Are events delivered in order?","a":"Yes, per subscription with one worker (the default): its worker takes the queue in order. Workers(n) trades order for throughput."},{"q":"How fast is it?","a":"Publishing to an Async subscriber costs about 59 ns, to a Sync one 36 ns, with no allocation (events.md, measured against kelindar/event, cskr/pubsub, EventBus and watermill)."},{"q":"What happens to queued events at Close?","a":"They are handled: Close closes the queues to new events, the workers drain them, and Close waits for that up to its context's deadline; past it, Close says how many were left."},{"q":"Which policy should I pick?","a":"Block when every event matters and the publisher may wait; DropOldest when only recent events matter; DropNewest when the first ones matter; Reject when the publisher should decide."}]}); |