Taming reactive node.js: Stream-oriented architecture with Nest

The stream-oriented architectures are still very rare in the node.js world. While reactive programming becomes a first-class citizen in the front-end applications, we don’t have too many materials on how to apply some of these concepts in our server-side apps. In this talk, Kamil will show you Nest framework as well as introduce you to the world of event-driven systems, CQRS, and Event Sourcing concepts (and how to really take advantage of them).

1. KAMMYSLIWIEC TAMING REACTIVE NODE.JS: STREAM-ORIENTED ARCHITECTURE WITH NEST
2. KAMMYSLIWIEC KAMIL MYSLIWIEC NESTJS CREATOR | CO-FOUNDER OF TRILON.IO
 @KAMMYSLIWIEC

3. KAMIL MYŚLIWIEC SOFTWARE ENGINEER
 @KAMMYSLIWIEC

4. KAMMYSLIWIEC Adonis Express Koa Nest Fastify
5. KAMMYSLIWIEC NESTJS WWW.NESTJS.COM
 @NESTFRAMEWORK

6. KAMMYSLIWIEC STREAM-ORIENTED
 ARCHITECTURE
7. KAMMYSLIWIEC 1 2 3 X
8. THINKING REACTIVELY
9. KAMMYSLIWIEC ADVANTAGES
10. KAMMYSLIWIEC READABILITY
11. KAMMYSLIWIEC LOOSELY COUPLED
 ARCHITECTURE
12. SELF-DESCRIBABLE
13. KAMMYSLIWIEC DISADVANTAGES
14. KAMMYSLIWIEC COMPLEXITY
15. KAMMYSLIWIEC REDUX
16. EXAMPLE
17. KAMMYSLIWIEC
18. KAMMYSLIWIEC https://localhost:3000/dragons https://localhost:3000/dragons/:id https://localhost:3000/heroes https://localhost:3000/heroes/:id https://localhost:3000/heroes/:id/slay-dragon
19. KAMMYSLIWIEC export interface SlayDragonDto { dragonId: string; }
20. KAMMYSLIWIEC LAYERED 
 ARCHITECTURE
21. KAMMYSLIWIEC CONTROLLER DATA
 ACCESS SERVICE
22. KAMMYSLIWIEC @Patch(‘:heroId/slay-dragon‘) async slayDragon( @Param('heroId') heroId: string, @Body('dragonId') dragonId: string, ) { await this.heroesService.slayDragon(heroId, dragonId); }
23. KAMMYSLIWIEC @Patch(‘:heroId/slay-dragon‘) async slayDragon( @Param('heroId') heroId: string, @Body('dragonId') dragonId: string, ) { await this.heroesService.slayDragon(heroId, dragonId); }
24. KAMMYSLIWIEC @Patch(‘:heroId/slay-dragon‘) async slayDragon( @Param('heroId') heroId: string, @Body('dragonId') dragonId: string, ) { await this.heroesService.slayDragon(heroId, dragonId); }
25. KAMMYSLIWIEC SERVICE
26. KAMMYSLIWIEC const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; await Promise.all([ this.heroesRepository.save(hero), this.dragonsRepository.save(dragon), ]);
27. KAMMYSLIWIEC const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; await Promise.all([ this.heroesRepository.save(hero), this.dragonsRepository.save(dragon), ]);
28. KAMMYSLIWIEC const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; await Promise.all([ this.heroesRepository.save(hero), this.dragonsRepository.save(dragon), ]);
29. KAMMYSLIWIEC const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; await Promise.all([ this.heroesRepository.save(hero), this.dragonsRepository.save(dragon), ]); TRANSACTION
30. KAMMYSLIWIEC MORE FEATURES
31. ARTIFACTS
32. KAMMYSLIWIEC hero.experience += dragon.level * 100; dragon.isAlive = false; if (dragon.level >= 3) { const item = await this.itemsService.getRandomItem(); hero.items = hero.items.concat(item); }
33. COINS
34. KAMMYSLIWIEC hero.experience += dragon.level * 100; dragon.isAlive = false; if (Math.random() < 0.3) { hero.coins += Math.floor(Math.random() * 101) + 100; }
35. KAMMYSLIWIEC if (dragon.level >= 3) { const item = await this.itemsService.getRandomItem(); hero.items = hero.items.concat(item); if (Math.random() < 0.3) { hero.coins += Math.floor(Math.random() * 101) + 100; } }
36. KAMMYSLIWIEC STREAM-ORIENTED ARCHITECTURE
37. KAMMYSLIWIEC @Patch(‘:heroId/slay-dragon‘) async slayDragon( @Param('heroId') heroId: string, @Body('dragonId') dragonId: string, ) { await this.heroesService.slayDragon(heroId, dragonId); }
38. KAMMYSLIWIEC @Patch(‘:heroId/slay-dragon‘) async slayDragon( @Param('heroId') heroId: string, @Body('dragonId') dragonId: string, ) { await this.commandBus.execute( new SlayDragonCommand({ heroId, dragonId }), ); }
39. KAMMYSLIWIEC COMMAND
 HANDLER
40. KAMMYSLIWIEC @CommandHandler(SlayDragonCommand) export class SlayDragonHandler implements ICommandHandler<SlayDragonCommand> {}
41. KAMMYSLIWIEC @CommandHandler(SlayDragonCommand) export class SlayDragonHandler implements ICommandHandler<SlayDragonCommand> {}
42. KAMMYSLIWIEC async execute(command: SlayDragonCommand) { const { heroId, dragonId } = command; const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; }
43. KAMMYSLIWIEC async execute(command: SlayDragonCommand) { const { heroId, dragonId } = command; const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; }
44. KAMMYSLIWIEC async execute(command: SlayDragonCommand) { const { heroId, dragonId } = command; const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.experience += dragon.level * 100; dragon.isAlive = false; }
45. KAMMYSLIWIEC async execute(command: SlayDragonCommand) { const { heroId, dragonId } = command; const [hero, dragon] = await Promise.all([ this.heroesRepository.findOneById(heroId), this.dragonsRepository.findOneById(dragonId), ]); hero.slayDragon(dragon); hero.commit(); }
46. KAMMYSLIWIEC slayDragon(dragon: Dragon) { this.experience += dragon.level * 100; dragon.isAlive = false; this.apply(new HeroSlayedDragonEvent({ heroId: this.id, dragonId: dragon.id, dragonLevel: dragon.level, expPoints: this.experience, })); }
47. KAMMYSLIWIEC slayDragon(dragon: Dragon) { this.experience += dragon.level * 100; dragon.isAlive = false; this.apply(new HeroSlayedDragonEvent({ heroId: this.id, dragonId: dragon.id, dragonLevel: dragon.level, expPoints: this.experience, })); }
48. KAMMYSLIWIEC COMMAND COMMAND
 HANDLER EVENT EVENT EVENT EVENT
 HANDLER EVENT
 HANDLER EVENT
 HANDLER
49. KAMMYSLIWIEC EVENT
 HANDLER
50. KAMMYSLIWIEC @EventsHandler(HeroSlayedDragonEvent) export class HeroSlayedDragonHandler implements IEventHandler<HeroSlayedDragonEvent> {}
51. KAMMYSLIWIEC @EventsHandler(HeroSlayedDragonEvent) export class HeroSlayedDragonHandler implements IEventHandler<HeroSlayedDragonEvent> {}
52. KAMMYSLIWIEC async handle(event: HeroSlayedDragonEvent) { // persistence logic }
53. KAMMYSLIWIEC async handle(event: HeroKilledDragonEvent) { const { heroId, dragonId, expPoints } = event; await Promise.all([ this.heroesRepository.update(heroId, { experience: expPoints }), this.dragonsRepository.update(dragonId, { isAlive: false }), ]); }
54. KAMMYSLIWIEC async handle(event: HeroSlayedDragonEvent) { // business logic await this.eventsStore.save(event); }
55. KAMMYSLIWIEC MORE FEATURES
56. KAMMYSLIWIEC COMMAND COMMAND
 HANDLER EVENT EVENT EVENT EVENT
 HANDLER EVENT
 HANDLER EVENT
 HANDLER SAGAS
57. ARTIFACTS
58. KAMMYSLIWIEC @Saga() dropItem = (events$: Observable<IEvent>) => events$.pipe( ofType(HeroSlayedDragonEvent), filter(({ dragonLevel }) => dragonLevel >= 3), map(({ heroId }) => new DropItemCommand({ heroId })), );
59. KAMMYSLIWIEC @Saga() dropItem = (events$: Observable<IEvent>) => events$.pipe( ofType(HeroSlayedDragonEvent), filter(({ dragonLevel }) => dragonLevel >= 3), map(({ heroId }) => new DropItemCommand({ heroId })), );
60. KAMMYSLIWIEC @Saga() dropItem = (events$: Observable<IEvent>) => events$.pipe( ofType(HeroSlayedDragonEvent), filter(({ dragonLevel }) => dragonLevel >= 3), map(({ heroId }) => new DropItemCommand({ heroId })), );
61. KAMMYSLIWIEC @Saga() dropItem = (events$: Observable<IEvent>) => events$.pipe( ofType(HeroSlayedDragonEvent), filter(({ dragonLevel }) => dragonLevel >= 3), map(({ heroId }) => new DropItemCommand({ heroId })), );
62. COINS
63. KAMMYSLIWIEC @Saga() dropCoins = (events$: Observable<IEvent>) => merge( events$.pipe(ofType(HeroSlayedDragonEvent)), events$.pipe(ofType(HeroFoundItemEvent)), ).pipe( mergeMap(({ heroId }) => Math.random() >= 0.3 ? empty() : of(new DropCoinsCommand(heroId)), ), );
64. KAMMYSLIWIEC @Saga() dropCoins = (events$: Observable<IEvent>) => merge( events$.pipe(ofType(HeroSlayedDragonEvent)), events$.pipe(ofType(HeroFoundItemEvent)), ).pipe( mergeMap(({ heroId }) => Math.random() >= 0.3 ? empty() : of(new DropCoinsCommand(heroId)), ), );
65. KAMMYSLIWIEC COMMAND COMMAND
 HANDLER EVENT EVENT EVENT EVENT
 HANDLER EVENT
 HANDLER EVENT
 HANDLER SAGAS
66. KAMMYSLIWIEC OPEN SOURCE
67. KAMMYSLIWIEC
68. KAMMYSLIWIEC WWW.TRILON.IO
69. KAMMYSLIWIEC THANK
 YOU @KAMMYSLIWIEC