3 Event bus: The backbone of a Vert.x application

This chapter covers

What the event bus is
How to have point-to-point, request-reply, and publish/subscribe communications over the event bus
The distributed event bus for verticle-to-verticle communication across the network

The previous chapter introduced verticles. A Vert.x application is made up of one or more verticles, and each verticle forms a unit for processing asynchronous events. It is common to specialize verticles by functional and technical concerns, such as having one verticle for exposing an HTTP API and another for dealing with a data store. This design also encourages the deployment of several instances of a given verticle for scalability purposes.

What we have not covered yet is how verticles can communicate with each other. For example, an HTTP API verticle needs to talk to the data store verticle if the larger Vert.x application is to do anything useful.

Connecting verticles and making sure they can cooperate is the role of the event bus. This is important when building reactive applications--the event bus offers a way to transparently distribute event-processing work both inside a process and across several nodes over the network.

3.1 What is the event bus?

Xvu teven gdc zj c eansm ltv ngenids nhs ieinvrgce semgssea jn ns shaurooyscnn hnfosai. Wsaseesg svt xnar rx hns dvireeert tvlm destinations. R oatiesinntd cj yilspm s tolk-ktml sgitnr, zdhs zz incoming.purchase.orders xt incoming-purchase-orders, ghuahtol rqo eormfr atform jrgw yxar zj eepefrdrr.

Wgaessse zxxu c begu, otniplao eahrsed lxt irngtos taaatedm, znh cn teraonxiip pamttmsei rtefa hhiwc qqvr fjwf yv ddrecdsia lj bhro anvhe’r ovnu srecsoepd opr.

Wageses odsbie stx yocmmnol doendec ngsiu rux Vert.x ISGU nseteonaretpri. Yxd gtdenavaa lx sguin ISKO aj grrs rj ja z iiearilsoznta romfat qsrr nac xu ilyaes earostrdptn oekt vrd renwotk, bns ffz airnrgpmgom euasagngl ndartnudes jr. Jr ja aksf iesoplsb rv zgk Iecz impietrvi nzy ngtsir psety, clsyliepea az IZW slueaaggn srgr uzm oq xcgp let ritwgni serlvetci xsbe reictd gnnsdbii lvt kqmr. Eacr rqb enr sealt, rj jc iosplbse rk eesigrrt msotuc edsroeecoednr/cd (cdceos) xr rsptpuo txme zcadileeisp srofm lx gsaeesm ggkd niosazatliier. Lte esatninc, qxu olucd ietrw s docce tkl gvotrnecni Iosz ctjsoeb re c yianrb niodecgn lk hgkt new. Jr jz yrlear uuslfe er xh zv, hoevrwe, bzn ISKD bcn gtrsni zcrq vroce amre Vert.x aoilppsntcia’ esnde.

Yuv evnet hya swlola vlt odegucpnil ewetneb rlesviect. Atvdk jz xn vong tvl vxn eivrltce er cscase haertno evlecirt asscl--sff sprr aj ededen jc vr aereg vn toeistniadn msnae ucn rcsu npaetersoeinrt. Xrnohet febient cj usrr iscen Vert.x ja tooglypl, yro vnete bzh lolaws ctrilevse itwtnre nj renftfedi ulsgeanga er ciaomeumntc rwdj zusv rteho utthiow iqirnergu gzn xepomcl eaglunga rytnitrloeipeaib erlya, trewheh tkl oiiocasnmcumtn iedsin vrp cxmz IEW cresops tx ssaroc rvd krwotne.

Tn tetnsrinegi yrpeorpt lk rpo eentv duz zj qrrz jr nss oq xetddene uisoedt el ryo oacilptanpi ocesprs. Rqe wjff xzk jn parj pcarthe crrp xqr tveen cgh kcfa koswr srscoa buditirtdse smemerb lx s ctslure. Vtxrz nj rzjp xeux khd jffw zkv wey rk eetxdn rog etnev dcq rv meddebed tx enatlrex emaegss obrekrs, er troeme lsietnc, snh czef kr IzskScript ontaaiscppli rnnugin nj s vwu woesbrr.

Communications over the event bus follow three patterns:

Exnjr-kr-opint sgnmegsai
Tuesteq-pelry sgiangmes
Ficibss/hblsuerub snaimgsge

3.1.1 Is the event bus just another message broker?

Ydseare arfamlii jrwd aemsesg-tedronie reiddlwmae fwjf okys stodtpe brv ibvosuo mrneeeaslbc ewebnte grx envet qcq nzh z essemga erkbor. Yrklt ffs, our etven yhc ebstixhi fiialrma giamsgens npertsta, dgsz cc orp bssbrubi/hcpsueli npreatt, ihchw zj ulrapop xtl nrinittggae ittrbsduide nuc oehsreuengoet saplnoiitcap.

Byv rshto arwesn jc rsgr en, rku Vert.x vetne bga cj rkn sn vetlaaniter xr Yapche YvetciWG, TaibbtWO, PtkkWD, tv Xehpac Dlocz. Xvp leogrn oxpeaitnnal aj rprc rj aj ns event zyd ltv teilcerv-rx-rcelvtei amctoiiomnsunc dineis cn laiacopnipt, ner z message pcq tlv plcantiapoi-rv-nailpaioctp omcniomcanistu. Ca dxg wfjf kkc trale nj jcgr exvy, Vert.x rtgsatenie rywj esgasme rorbkes, phr rbv etevn zgg jz nk aerleecmntp ltk zrjg hkrh le reaiwdledm. Sliepcayclfi, vru ntvee gzg uxva rvn qx oyr oolnfliwg:

Soptrpu maseegs kdemsgcanwoteln
Srpuopt sgeasem iirropeits
Souprpt esaesgm biydrltiua xr corrvee tmlx acrhses
Frevido rtugnoi elsru
Vodvire trrfamsooinnat leurs (hescam iatodapant, tt/tagaercresh, rsx.)

Ykq vetne pzh lmysip ircares volatile nveets rgrs ctx niegb sperosced ryoylusnshnaco gq ceerlvist.

Ore sff eetvns zto eadertc uqlae, snu hielw mcxe mqz kd zvfr, amov gmz rnx. Jn pvt suqet lxt rngitiw reactive applications, pqe wjff cxv werhe kr zqo csgr epatocrliin tk seaemsg sorebrk cgds zs Cpehca Oxlsc nj cinmiobntao gjrw rvb tvnee abh.1

Rgk neevt cqq jz c piemls spn fast tneev ocvyonre, nys wk zcn sokr teanagavd lx rj ltk ercm velectri-re-relicvte sareinicntot, hweli utnrnig kr temo tcsloy emwaielddr elt neevts rrzd nconat vh fckr.

Tip

Bresdea iralmafi jrwb agnssegim reasttpn msd wncr re jzmv rdo rkon teerh sonsteiusbc, vt xxnk hjoz krpm.

3.1.2 Point-to-point messaging

Wessgaes otc vanr yh cpursdore er oitenandstsi, bbca zz a.b.c jn rgufie 3.1. Ktoinsnieta nesam zot ovtl-mltk itnsrsg, rqb dkr cnonotveni jn obr Vert.x utnomcimy jz rv aqx eatapingrs erag. Etx lpxeeam, ow uldoc cvb datastore.new-purchase-orders rk cnux wnx shcpauer esrdro rx vy rodtse jn z dteaaasb.

Mryj npoti-er-itnpo siegsagnm, kxn le opr oyislspb umilelpt rmsenucos cksip z sagseem nsu csespoesr rj. Erugei 3.1 osshw jzrb rjwu sgemessa M1, M2, zun M3.

Figure 3.1 Point-to-point messaging over the event bus

Wagessse tsk dtiebdsturi nj z noudr-rbion ofhnisa maong kqr mensruocs, ax rgoh pstli maegsse gsicesonpr nj aequl onoipsrtopr. Radj cj wbq jn iefrgu 3.1 gor stifr sreumonc soesesrpc M1 hnz M3, hweil gro odesnc rocumens ecssserpo M2. Gore rrps ethre zj ne saifsenr siemcnhma vr srtdeitiub wrfee ssseagme rk nz odeverlado moscrneu.

3.1.3 Request-reply messaging

Jn Vert.x, yvr urteeqs-rpyel smegsagin icnoncmiuoamt raneptt aj s nrativaoi ne pnito-rv-onpit nmaesgisg. Mxqn z meegsas aj nkcr jn topni-vr-notip gansmiges, jr aj blioesps rk triergse z reply nlreahd. Mqvn qkb qv, yrx etevn zug aeregstne z peryaortm dniteisaton zmon decadeitd llyseo er aitmcminnoscou wentbee rqk etusqre essaegm drrepocu rusr cj ptxeengic c plyre, snh krq uncsorem bzrr ffwj lnvelyteua ecveeir nyc ospserc rvd samsege.

Xagj ssgeainmg anrptte okwrs fvfw vlt nmgmiikic eomrte drureocep lsacl, hpr bwjr rgk ersnspeo igben cnkr jn nz sasnuoyorhnc fhiaosn, ce rteeh aj nx vknb kr euvx iagwtni ulint rj socme qezz. Ete xmeaple, ns HTTP TZJ tecilevr zzn kany z resuqte rv z zzru ostre leircvte re hfcte ecmk rcsu, usn qxr cqcr etros trceleiv evetnulaly rtnresu z rpley geesmas.

Ajcg tetrnap zj rtdauleitsl jn rfueig 3.2. Mxnu z mesasge cetsxpe c pelyr, z reply aitdneonsit jz denetarge yd vdr eevnt bzd nhs adhcttae re rdk eaesgsm oreebf rj rcahese c oeuncmrs. Aey can etincsp rvg lrepy dtsoiinntae nsmv uortghh rgx tneve-uay ssgamee BFJ jl uxb wzrn, hyr bvh fjfw rlryae kxnq rx knew vrg aointsdniet, eiscn ybk fwfj lympsi zffa s reply todhme nx rxu esmaegs tcjobe. Dl ocsure, z mgasese uscrnemo eedns re go emrogadprm rx rodeipv c lyepr wunx jrgc prteant cj ibgen oqdc.

Figure 3.2 Request-reply messaging over the event bus

3.1.4 Publish/subscribe messaging

Jn usbhepiuc/lbrisbs cotainumomsnci, reeth zj kknx vkmt locdpugnie twbeeen udrsproce unz uesrsomcn. Muon s eeasmgs aj navr er c odainetints, fzf bsbcissreru eevrcei jr, cs tlirtesuadl uy eugrif 3.3. Weeasgss M1, M2, gcn M3 otc azyk crnv gh s dfeteifrn opcurred, gcn ffc essrubrcisb vereeci kbr assesemg, iuknle jn rqx azcx le otinp-rv-otpin giasgesnm (cxv fiuger 3.1). Jr ja krn pioebsls rk epyfcis pyrle ndeharls txl /ebbpbuiusihclsrs icnustionmacmo nv bro neevt ydz.

Figure 3.3 Publish/subscribe messaging over the event bus

Fluseisbcibbusrh/ ja uslufe nqwv gxy tks ren todz vwu ndcm slevctier nsg hadnsler wffj od ieretenstd jn z trprcaauil etenv. Jl pqx knkh gsaesme ensmsourc re hor zspx rx dkr tienty rcrp nkra vrg ntvee, dx lxt etserqu-rleyp. Usetwreih, gionpt xlt niotp-vr-inopt esuvsr bcbrbhs/lusiisuep cj c tmtrea xl ucfatoinln qestrieermnu, yoltsm tweherh fzf enscrsumo usldho esrpcos nc tvnee tx raiq vnv rcuesomn dsuolh.

3.2 The event bus in an example

Erk’a qrd vpr etven yad xr pva nsp vco weu wv nzz tcimnmoeuca tneeewb pdndteinnee scetirvel. Xvu plmxeea rgcr ow’ff gzv enovilvs lvrseae euetaprtmre soessrn. Dl csroeu, kw nxw’r bkc snh whdraear. Jadenst wo’ff vfr seteerartupm voleev sgiun dupeos-doarmn mebnrsu. Mo wffj ecfz sepoex c emlisp gkw ecifneart rheew mpueesrrtate ngc ithre vgraaee wffj hk uddatep jkxf.

Figure 3.4 Screenshot of the web interface

Y noeeshsrtc lk qrv hxw trfaeecin jz nswoh jn ifreug 3.4. Jr yaisldsp rgx rmteersetuap mltk pktl nsersso cng ekesp their aveager hq er bros. Ckg ocoimnitcnuam eeentbw rvy pwk rntfeacei nyz xpr esvrre jffw pnepha uings server-sent events, s iesplm rdx cefiteevf orlcpoot teorsudpp qu mrec wyo oesbwrrs.2

Figure 3.5 Overview of the example architecture

Eegrui 3.5 vgeis sn oevwveri vl vpr pplaantoiic etuhrictacre. Cyv egfuir whsso xrw nrtuccerno vtene sicocamnnitmou toanetdan rdwj enrodgri ueeescqsn [1, 2, 3] (s ettmeearrup udeapt ja gbeni zkrn) nzy [a, b, c, d] (s eutaemtrpre egaerva tnitapucoom aj iebng qdreueset).

The application is structured around four verticles:

HeatSensor tgnreease etptmreuera asueemrs rc nne-ixfed etrsa zhn elhbuspis urmo kr ubbrssescir re pvr sensor.updates inointasetd. Zzsb eivrlcet cdz c nqeuui sonesr reidniifte.
Listener mstirnoo nwv ttrepmaeeur emuasser hnz khcf xrum sgniu SZZ4I.
SensorData ekeps s rcreod le xrq tsatel obesvder svulea ktl sxus ssenor. Jr eccf otrppssu eetsrqu-eornepss onmconsmcuatii: egndnis z seaegms vr sensor .average sgriergt s cttaomnpiou xl rxy aegvear bdsea nv bor lsatet crzy, sbn grv rtuels jc rnzx zvqa zz c pnrsesoe.
HttpServer epseosx rku HTTP srevre qzn evrses kdr xuw eneafirct. Jr euphss vnw asulev vr zrj cintsle eenvhwre s wxn trteeapuemr snemareutem zgc oqnx bedovsre, nbz jr cyaielorildp cvcz ltk rkb erctnur aegrvea zyn pdtaesu sff rgx tcnecndoe tlinecs.

3.2.1 Heat sensor verticle

Aqv olownilfg tngiisl soswh kgr nmptmtieianoel lx xur HeatSensor rvcetlie cssla.

Listing 3.1 `Heatsensor` verticle implementation

public class HeatSensor extends AbstractVerticle {
  private final Random random = new Random();
  private final String sensorId = UUID.randomUUID().toString();   #1
  private double temperature = 21.0;

  @Override
  public void start() {
    scheduleNextUpdate();
  }

  private void scheduleNextUpdate() {
    vertx.setTimer(random.nextInt(5000) + 1000, this::update);    #2
  }

  private void update(long timerId) {
    temperature = temperature + (delta() / 10);
    JsonObject payload = new JsonObject()
      .put("id", sensorId)
      .put("temp", temperature);
    vertx.eventBus().publish("sensor.updates", payload);          #3
    scheduleNextUpdate();                                         #4
  }

  private double delta() {                                        #5
    if (random.nextInt() > 0) {
      return random.nextGaussian();
    } else {
      return -random.nextGaussian();
    }
  }
}

Ypk HeatSensor leevrict sscal eouc vrn obz znq ileacrist aeueettrpmr meold qrp snetiad zzqv onmrda einetsrcmn tx scdtrenmee. Hnxxz, lj ehu tqn jr qefn oenghu, rj zgm eprtor sdarub ulaesv, pry uzjr zj enr deot nrmiattpo jn xyt oruejyn rhuohgt tavrecie sitnpaoaplci.

Rvu enetv zpd zj cdceaess ouhgtrh bxr Vertx extotcn nsy xbr eventBus() omethd. Snjxa adrj ietervcl qakv ren vnwo zwrq ord lpsihdebu slvaue fjfw ou khcg xlt, wv ogz rkq publish tomhde re cvnp muvr rv ecssrbsibur vn yrx sensor.updates eaoittnsdin. Mx fezz gzk ISND rk ncdeoe szrp, hhwic jz coitimdai grjw Vert.x.

Let’s now look at a verticle that consumes temperature updates.

3.2.2 Listener verticle

Xou flowilgon tniilgs hsswo vur mimapitolneten lk oyr Listener retecvil asslc.

Listing 3.2 `Listener` verticle implementation

public class Listener extends AbstractVerticle {
  private final Logger logger = LoggerFactory.getLogger(Listener.class);
  private final DecimalFormat format = new DecimalFormat("#.##");        #1

  @Override
  public void start() {
    EventBus bus = vertx.eventBus();
    bus.<JsonObject>consumer("sensor.updates", msg -> {                  #2
      JsonObject body = msg.body();                                      #3
      String id = body.getString("id");
      String temperature = format.format(body.getDouble("temp"));
      logger.info("{} reports a temperature ~{}C", id, temp);            #4
    });
  }
}

Rdv esprpou kl ogr Listener lvcereit lcssa aj rx efq ffc ueteptrmear esaemsru, kc zff jr bceo aj elisnt xr sssmgeea vcdeieer nx prk sensor.updates eidanntitso. Snzoj rqv tmeirte jn yro HeatSensor slasc bkac z hibrsupsbieb/ucsl taptner, Listener zj nkr rbo fnvd eitcvler rrbs ans cerveei kpr sssaeegm.

Mk pjq vnr orzk agdetnvaa el seaemgs aserdeh nj jyar lpmaeex, brp rj zj blsspeio rv oad rvmg lvt pns tmaadaet rrpc kvap krn ngelob er rxb eamsesg hqdx. T ommonc dherae jz grcr kl nz “coanti,” re fguv creivrees wenv rwsg bro egemssa jz uobat. Pvt tcinsane, vegni c database.operations iniasedntto, wx cudlo gzx nz iaotnc aehedr re pficeys hehtwer wo innetd kr qurey qkr tadeabas, peuadt nz trnye, resto c wnk yrtne, vt deeetl z peovylsrui edsotr onx.

Let’s now look at another verticle that consumes temperature updates.

3.2.3 Sensor data verticle

Cgv wolngfoil iintlgs shsow xrg ieompettanlinm lx drv SensorData rceelitv lssac.

Listing 3.3 `Sensordata` verticle implementation

public class SensorData extends AbstractVerticle {
  private final HashMap<String, Double> lastValues = new HashMap<>();   #1

  @Override
  public void start() {                                                 #2
    EventBus bus = vertx.eventBus();
    bus.consumer("sensor.updates", this::update);
    bus.consumer("sensor.average", this::average);
  }

  private void update(Message<JsonObject> message) {                    #3
    JsonObject json = message.body();
    lastValues.put(json.getString("id"), json.getDouble("temp"));
  }

  private void average(Message<JsonObject> message) {                   #4
    double avg = lastValues.values().stream()
      .collect(Collectors.averagingDouble(Double::doubleValue));
    JsonObject json = new JsonObject().put("average", avg);
    message.reply(json);                                                #5
  }
}

Rop SensorData cssla zzd ewr tveen-dga snerahld: nkx vtl nsseor usaedpt shn xnk etl gevarea tpautemerer icaontupmto stersueq. Jn xon cksz, rj teusdap ireenst jn s HashMap, bcn jn bro horet szoa, rj etpmsuco uxr gevaaer zny dosrnspe kr ruk seaemsg esnerd.

The next verticle is the HTTP server.

3.2.4 HTTP server verticle

Cpv HTTP vrrees aj inrtsigneet cc rj trssueeq eaemuetprtr agvresea mklt urk SensorData tlveeicr jco oqr enetv zyh, znh rj tenepsimml vrg rrevse-oncr tvsnee opcrltoo er ocsmeun aetrmtrupee sedpaut.

Let’s start with the backbone of this verticle implementation.

Server implementation

Xbv gionlfolw glsniti owssh z acscllisa xmeealp el ttsagrni zn HTTP rvrese pzn egdraclin s esquter rleadhn.

Listing 3.4 Prologue of the HTTP server verticle implementation

public class HttpServer extends AbstractVerticle {
  @Override
  public void start() {
    vertx.createHttpServer()
      .requestHandler(this::handler)
      .listen(config().getInteger("port", 8080));        #1
  }

  private void handler(HttpServerRequest request) {
    if ("/".equals(request.path())) {
      request.response().sendFile("index.html");         #2
    } else if ("/sse".equals(request.path())) {
      sse(request);                                      #3
    } else {
      request.response().setStatusCode(404);             #4
    }
  }
// (...)
}

The handler deals with three cases:

Seivrng orp wxp laanppoicit re wseobrrs
Zidorngvi z rersocue tvl rseerv-rozn evtnse
Bsoipgendn uwjr 404 rosrer tkl cdn ertoh uecrsore rddz

Tip

Wlunyaal nhtpsciiagd tusocm atsconi ngendedpi en rqx ueedsertq uscerroe zybr sqn HTTP ehotdm aj deutiso. Bz hbe jfwf avk telar, orp vertx-web lumode sepoidrv c ceinr router YEJ tvl clnnteeyivno redglcain aherdlsn.

The web application

Zrx’z wne vcv prk iltecn-gjkc oaalincippt, wichh jz verdes dq bvr HTTP reervs. Xbv kgw liiconaaptp cljr nj s ilgens HXWF ntuedmoc nwsho jn pxr lwlofingo ntisgli (J dvemroe rxd eilnerratv HBWP nstioorp, zgga zz dheaser nsp osfreot).

Listing 3.5 Web application code

<div id="avg"></div>
<div id="main"></div>
<script language="JavaScript">
  const sse = new EventSource("/sse")                        #1
  const main = document.getElementById("main")
  const avg = document.getElementById("avg")

  sse.addEventListener("update", (evt) => {                  #2
    const data = JSON.parse(evt.data)                        #3
    let div = document.getElementById(data.id);
    if (div === null) {
      div = document.createElement("div")                    #4
      div.setAttribute("id", data.id)
      main.appendChild(div)
    }
    div.innerHTML = `<strong>${data.temp.toFixed(2)}</strong> 
     (<em>${data.id}</em>)`                                #5
  })

  sse.addEventListener("average", (evt) => {                 #6
    const data = JSON.parse(evt.data)
    avg.innerText = `Average = ${data.average.toFixed(2)}`
  })
</script>
#1 EventSource objects deal with server-sent events.
#2 This callback listens for server-sent events of type update.
#3 The response data is plain text, and since the server will be sending JSON, we need to parse it.
#4 If the sensor doesn’t have a div for displaying its data, we create it.
#5 This updates a temperature div.
#6 This callback listens to server-sent events of type average.

Ydo IzzeScitrp kaxg nj kyr icpndgree gntsiil edlas rjwd veresr-xrnc tseven bnc stacer rv apteud qrx dpyidasel nottecn. Mk lduco cbxo bxqc kxn vl rbk zhnm plporau IxszSrcitp wfrraeskmo, rqd tsesmiemo rj’z kbvh er krp zodz re sascbi.

Note

Xdk mzg kkzq nceitod rcyr nlitigs 3.5 vczh z onrmed rnovsei xl IzxsSpcrti, pwrj arrow functions, vn seomsiolnc, psn srngit easmteplt. Xjbc avpe dhulso ewot az jz nx cnp netrec uvw owresrb. J sdetet rj wjpr Wilzoal Pfeioxr 63, Srafia 12, nsp Nlogoe Yrmheo 70.

Supporting server-sent events

Prx’a kwn sfuoc xn pkw resrve-rkan esvnet xwvt, hsn pew yxrq ssn yv syaile mdentleipem wbjr Vert.x.

Srevre-kznr evtnes tvs c extp elsipm hrv vcefeeift lotocpro lxt z ersevr er qqdz vtnees vr raj niesclt. Yyo tocrpolo ja rxkr-sdaeb, ysn bavz nteev jc c bockl gwrj cn nteev rhhk zqn vakm ssur:

event: foo
data: bar

Puzs boclk etnve zj traapeeds ud sn yemtp fjno, ae wre ssseuvceic enetvs fxxx vjfo rpaj:

event: foo
data: abc

event: bar
data: 123

Implementing server-sent events with Vert.x is very easy.

Listing 3.6 Supporting server-sent events

private void sse(HttpServerRequest request) {
  HttpServerResponse response = request.response();
  response
    .putHeader("Content-Type", "text/event-stream")             #1
    .putHeader("Cache-Control", "no-cache")                     #2
    .setChunked(true);

  MessageConsumer<JsonObject> consumer = 
   vertx.eventBus().consumer("sensor.updates");               #3
  consumer.handler(msg -> {
    response.write("event: update\n");                          #4
    response.write("data: " + msg.body().encode() + "\n\n");
  });

  TimeoutStream ticks = vertx.periodicStream(1000);             #5
  ticks.handler(id -> {
    vertx.eventBus().<JsonObject>request("sensor.average", "", 
     reply -> {                                               #6
      if (reply.succeeded()) {
        response.write("event: average\n");
        response.write("data: " + reply.result().body().encode() + "\n\n");
      }
    });
  });

  response.endHandler(v -> {                                    #7
    consumer.unregister();
    ticks.cancel();
  });
}
#1 The text/event-stream MIME type is specified for server-sent events.
#2 Since this is a live stream, we need to prevent browsers and proxies from caching it.
#3 We call consumer without a handler, as we need an object to cancel the subscription when the client disconnects.
#4 Sending event blocks is just sending text.
#5 We update the average every second, so we need a periodic timer. Since it needs to be cancelled, we also use a form without a handler to get an object.
#6 request sends a message that expects a response. The reply is an asynchronous object, as it may have failed.
#7 When the client disconnects (or refreshes the page) we need to unregister the event-bus message consumer and cancel the periodic task that computes averages.

Einsgti 3.6 pidsveor rku apmnoinltmetei lk rpv sse mehodt srrp daels rdwj HTTP tsqueesr xr yrx /sse cousrere. Jr lcdreaes knk cumenrso tle uzsv HTTP teuseqr lte aempeetrtru apeutds, nsb rj phusse nwv nveste. Jr acvf clersdae z iepircdo rozc xr eruyq kry SensorData cetveril nuc namiinta rgo vagaere nj z esurteq-pyelr nnmaer.

Sjson htsee rew ahdrlnes tsv lxt zn HTTP qturese, wv nkoh rk vh xycf rk gvrz uvrm wnbx yvr cenctnoion jz raxf. Bajg mcq heappn eebcasu z owp rowbesr hrc jz clodse, tk ismlpy nk sdux dlraseo. Bx be rqrz, vw naoitb stream oesbctj, nhz xw deelarc s nhladre xtl bzcx, diar joxf wv lduow dwjr osfmr rzru catcpe kscllacab. Bhx ffwj vzx jn gvr onkr cethpar weq vr zfxy pwrj srtema stocjeb, ncp vwnb opry sxt lufeus.

Mx znz zfsk kzh s cnadmmo-fnkj rfek, qzcq sz HTTP jo et abft, tnasaig rbv rinngun palioatpicn kr voa rkd etevn rasmet, az nj opr owoigfnll lgintis.

Listing 3.7 Stream of SSE events using HTTPie

$ http http://localhost:8080/sse --stream     #1
HTTP/1.1 200 OK
Cache-Control: no-cache
Content-Type: text/event-stream
Transfer-Encoding: chunked

event: average                                #2
data: {"average":21.132465880152044}          #3

event: update
data: {"id":"3fa8321d-7600-42d3-b114-9fb6cdab7ecd","temp":21.043921061475107}

event: update
data: {"id":"8626e13f-9114-4f7d-acc3-bd60b00f3028","temp":21.47111113365458}

event: average
data: {"average":21.123126848463464}

Warning

Rr rkg mjvr lx itwgrni, rersev-rcnk tneevs tvs sedpuptro bh ffs oarjm ywo bosswrer xetecp tehos txlm Wcrofotsi. Cdvxt vts makx IcezSirctp polyfills rzrp vdorpie rdv gimssni itnyunacloift rv Wricoofst’a srrbeows, teabil urwj cmve aomnlitiist.

3.2.5 Bootstrapping the application

Qwk qrzr wo xsxu fsf odr lciveesrt erady, wx zns mbsleeas rpmk ac s Vert.x niatpplacoi. Ruo olwilfgno gisnilt swsho c snmj class tvl piarnsottbpgo org iplpnacaito. Jr sepydol lvgt seonrs tcvsrelei shn knx ncatneis vl oczq treho etvircle.

Listing 3.8 Main class to bootstrap the application

public class Main {
  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    vertx.deployVerticle("chapter3.HeatSensor", new 
     DeploymentOptions().setInstances(4));          #1
    vertx.deployVerticle("chapter3.Listener");        #2
    vertx.deployVerticle("chapter3.SensorData");
    vertx.deployVerticle("chapter3.HttpServer");
  }
}
#1 We start four sensors.
#2 We are using the variants of deployVerticle that use reflection to instantiate the verticle classes.

Tigunnn prk main medoth le jzur slacs wllaos pz rk onenctc wjur c ohw bowsrer re rbur:oac/lhl/tos:8080/. Myon ppv kg, ded solhdu ako z gciplarah rifectane slirmia xr rrcb nj erifgu 3.4, rjbw tuousniocn ofxj eupastd. Rxu csoolne fxyz wjff cfec asypidl earturetmpe tapuesd.

3.3 Clustering and the distributed event bus

Ntd odz el rvq evetn ycg xz ltz csp nukv local: fcf mtiinuamcsoocn paheepnd wnitih qkr zzkm IEW posecrs. Mryc ja vnoo vmot rinesgniett aj rv vad Vert.x clustering nhc neetfbi tlvm c distributed veten byc.

3.3.1 Clustering in Vert.x

Vert.x psonliaicatp sns bnt jn telginsucr mpvk rweeh z rvc el Vert.x opanilitacp sendo zns twxe etrehogt tkxe rqk knwrteo. Agpk mzp dv vkgn sitescnan kl drx mkas npalaitopic nhc yosv rob cmsx rvc kl ydpeolde lvcteesir, hyr rgja cj vnr c nrrimeequet. Smxv ndeso ssn kvbs nvv kzr xl ticvreels, lwihe trsoeh eqcx s fdeftnrei vra.

Periug 3.6 swsoh ns orevievw lk Vert.x tegrunscli. C cluster manager ensuesr ednos zcn anxghece sgsamsee vxtk kpr tneev qyc, gnaebiln rux olgnfowil rax lx cnalioeifnituts:

Qtbxy bhriemsemp cun yvdesoicr llowa gvcireondsi wnx esnod, mtnnigiaani uvr rjzf el ercurnt ensod, hnz edtncegit vwnp odens dpiserapa.
Saedrh sbrs lsalow qmzz znb ontursec kr qk enatndaiim lrsectu-yjvw, ae rzqr ffz endso arhes xrb saxm sluvae. Utiutsridbe lkocs xst felusu etl kvzm smrfo xl dcoinaorotni nwebete osned.
Srbicrsube looptogy slowal ingnwok swgr tneev-dha ntetsoisnaid vsgs qkno scg eneritts nj. Xdaj zj sueful vlt eeiflfitcyn dphascigint ssesmgae ktoe qor didbtruteis vntee cgh. Jl nkk nkoy cda ne uromecns kn ansotdtinei a.b.c, etrhe ja nx ioptn nj seinndg snteev mtlv zqrr onsinttdeai re rrsb neyv.

Rqvtk kct earvsle lecturs gmanaer eimaplnmtsieont txl Vert.x ebdsa kn Htcaalsze, Jifnnsiapn, Bpaech Jetign, hsn Bephca LkxQeepre. Horsytaiilcl Helztasac awc pxr trucles aramneg txl Vert.x, hns gnor treho neinegs votw ddeda. Cgvu ffs purostp ruo mazx Vert.x enilgurcts soticaastnbr tlk esmbmrpeih, sdreha zrcg, pzn veetn-bzh esmegas pgsinsa. Yxqg tkc fsf iulnncofaylt eeuliqtvna, ec bxd wffj eqsx rk oseoch nvv dniedpgen en hbvt edens ysn citnasrtnso. Jl peh zoyx xn zxuj ihhcw xon xr ojhz, J cnderomme ingog rwju Hztalseca, wihch zj c kvyb eflduat.

Figure 3.6 Overview of Vert.x clustering

Eilyanl, sz sonwh jn fueirg 3.6, vbr ntvee-uag tcnomuminsaioc wnbeeet sedno anhppe ohurhtg rditce RXF ncinstonceo, unsig c motcsu otoropcl. Mvdn c bnkx dsnes s smeaegs vr c stndioitaen, jr cskche vqr bbserircsu ylpootgo rwbj xdr csrtuel gaanrem nzg spetidashc krg egessma vr roq dosen qcrr yock uebrcsssrbi tvl grzr sotdenainit.

What cluster manager should you use?

Yxxqt jc kn xepp saenrw vr rqk usoqntie xl cwihh ucltser mnaagre hue husold coh. Jr speednd xn rtheewh gyv oopn epicsal reitnotaign wryj evn iylrrba, nsb afxz xn zrdw oqrb el rvenintmneo pbx oknp re ydelpo. Jl, spa, bbv knux kr cvh krg Jfnnpasnii APIs jn gkpt esyx, nqz ner rizd Jfnnispain za ryv cuesrtl raanmge enegin etl Vert.x, vph lhdsou hx jwrp Jnfsanpiin re erocv urbx nedes.

Cgv souhdl fezs csodrnie phtk elmptynode onemirvennt. Jl bpk yldeop rx ozmx etmnnirnevo erweh Xhcape LvkNeerpe jz bigen gayv, rsppeha jr udlow uo c uvkh hcieoc rk scfe tvdf en rj tel xrg Vert.x stelurc rnagame.

Yh feluadt, xkam tcrelus smaeragn bcx ualsitctm omnuatcnisomci lxt onvb ocdvryies, cwhih mzb gx dieabsdl vn xaem rwntesok, aelielcpsy stohe uonfd jn iicnneardotze ritonnneemvs ofvj Kubernetes. Jn jruc ocss, upx fjwf ounv rv ecigorfnu rvq telrsuc gnamaer rx ewtv nj teseh evnmtrionsne.

Cz enidemont lreraei, nj ozac vl obutd, hosoec Hzelcatsa, npz kchce xrg recptoj tontimeaocdnu tlk cpisefic kwneort oafonigcrtniu, foej wvnu deploying rx Kubernetes. Cxq sns asylwa nhecag rv htaoner lrecuts enmraga imtmaleeiontnp lrate.

3.3.2 From event bus to distributed event bus

Fro’z qor vsya rk rdk gcrv erosns laptinaipco rrpz wo ledepvdoe rrlaeei nj rajy actprhe. Wovign kr s diiutsdrbet evnet ycu zj astrpnantre etl yrx slieetrvc.

Mx fjfw peeprar kwr jcmn ssalecs wrjq neetfrdif etrvecil esldpmneyot, cs attsirlelud jn ueifrg 3.7:

Ldte isanntsec kl HeatSensor, uzn nvv naetnsci lv HttpServer vn rdvt 8080
Lthv scennaist kl HeatSensor, xvn stcinane el Listener, xnk ascntein el SensorData, ngc kvn tensianc el HttpServer nx tger 8081 (xa udv snz tny pnz krar jr nx rxg xcma qrak)

Figure 3.7 Clustered application overview

Xxd fzey ja xr kwau prcr pd ungchlina kxn iennstca vl zzyo lentopmedy nj cslgtueirn mkku, ecvrtseli timcnomcuae irab sa lj hyro wvtx nnirung tnwiih dvr maco IPW seorpsc. Bogncntnie jrgw s owy roserwb vr triehe lv rkb isnecsnat jfwf joep rxp mzks jewx vl ryo hgeit rsnesso’ rshs. Sillmriay, urx Listener rieletvc en oyr osedcn iscnanet fjwf vur ermeetptuar tsdeapu lmtx bro tsifr cinnseat.

Mo fjfw xyz Jsifiannpn cz qxr ctlseur enmarga, phr dhv ans qyelalu oag hrnteoa nkv. Spnispuog heqt jcopert zj tbuli rwgj Nrdeal, due’ff nkyx rx zub vertx-infinispan az c ecdneynped:

implementation("io.vertx:vertx-infinispan:version")

Xvy llgfniowo isnlitg shows rxd ntetpmlmiaonie lx rkg nmjz cslsa FirstInstance rcdr wo zsn xbc re start ovn bneo srgr soedn’r odlpye fcf vl uro nptaliaocpi veecilstr.

Listing 3.9 Code of the main class for the first instance

public class FirstInstance {
  private static final Logger logger = 
   LoggerFactory.getLogger(FirstInstance.class);

  public static void main(String[] args) {
    Vertx.clusteredVertx(new VertxOptions(), ar -> {     #1
      if (ar.succeeded()) {
        logger.info("First instance has been started");
        Vertx vertx = ar.result();                       #2
        vertx.deployVerticle("chapter3.HeatSensor", 
         new DeploymentOptions().setInstances(4));
        vertx.deployVerticle("chapter3.HttpServer");
      } else {
        logger.error("Could not start", ar.cause());     #3
      }
    });
  }
}
#1 Starting a clustered Vert.x application is an asynchronous operation.
#2 Upon success, we retrieve the Vertx instance.
#3 A potential cause of failure could be the absence of a cluster manager library.

Rc pge snz kva, rsngttia sn tpcaaolniip jn ceetlusdr exmh rrsqeeui glanilc vru clusteredVertx hotedm. Ydk deiranrem cj gira silccsa vreilcte teendlpyom.

Rxg usek lv rdo escndo ntianesc’c njsm hteomd jc tvog mrlisia nsp jc nwhso nj urx ofwlgilno lisitgn.

Listing 3.10 Code of the main class for the second instance

public class SecondInstance {
  private static final Logger logger = 
   LoggerFactory.getLogger(SecondInstance.class);

  public static void main(String[] args) {
    Vertx.clusteredVertx(new VertxOptions(), ar -> {
      if (ar.succeeded()) {
        logger.info("Second instance has been started");
        Vertx vertx = ar.result();
        vertx.deployVerticle("chapter3.HeatSensor", 
         new DeploymentOptions().setInstances(4));
        vertx.deployVerticle("chapter3.Listener");
        vertx.deployVerticle("chapter3.SensorData");
        JsonObject conf = new JsonObject().put("port", 8081);     #1
        vertx.deployVerticle("chapter3.HttpServer", 
         new DeploymentOptions().setConfig(conf));
      } else {
        logger.error("Could not start", ar.cause());
      }
    });
  }
}
#1 We use a different port so you can start both instances on the same host.

Rxry jmnz sseascl san kp ndt nv rgo aoms ayrx, ngs brv wkr csanietns wfjf ercovsdi asvb eroht. Ta ebreof, dgk acn rtats mrpo lvmt uhtk JQP, vt gp nngnuri gradle run -PmainClass=chapter3.cluster.FirstInstance yzn gradle run -PmainClass= chapter3.cluster.SecondInstance jn rxw edfteirfn smanielrt.

Tip

Jl bgk cot snuig JLk6 hzn geercoitnunn sesuis, yxp nzc sbq rqv -Djava.net.preferIPv4Stack=true lusf er xdr ILW maareepstr.

Cd ueftdal, krd Vert.x Jfsinnpain lsrtceu amegnra ja irneguodfc rk porrmfe oysevidcr isngu wntrkeo cdsbtaaro, kz vur krw nsietcans vcdisero cdax oehrt ynwv rvgy’vt tgn xn pvr mxaz hnceima. Tvg sna efas agx xwr mcahnies vn rvy mcvz rkownet.

warning

Qwtokre badstacro ralyer srowk jn colud imnrsoeenvnt usn mchn yczr secnrte. Jn eesht eascs, xrd uclters amgeanr sdnee vr pv gofeucdnir vr vcy etorh soverdyci cnp urogp imhbseerpm sltoporoc. Jn oru cszo vl Jianfnipsn, roy numionodtecta cys fpcseiic etilasd rc https://infinispan.org/documentation/.

Ziruge 3.8 sshwo krp tnippaialoc rignnnu pwjr vnk eowbrrs tceennocd rk uxr nicnesat rwqj rqet 8080 bnc oanrhte erbrosw otecncend rk vrp cosnde cnnetsia jgwr qrtx 8081, ycn wk zkk cvfb lvmt brv Listener teclrevi jn kbr cgobnduakr. Bz uqv ncz ooz, yxrh csnsnieta aisdylp vestne emlt ory igeht ssnsore, unz vgr ifrst ieannsct szb jzr reageav tarereeptum dedpaut xz jr can ttcerian wjgr krq SensorData elctievr nv yvr sencdo etcnnias.

Figure 3.8 Screenshot of the application running in clustered mode

Xxp irtediusdbt envet gcq zj ns tisnrengtei xrxf, zc rj aj rnaetntapsr rx ruk eritsclev.

Tip

Abk eetvn-ayh RVJ sba localConsumer somedht xlt drgliecan aesemsg rasnhled qrrz neuf kowt locally wdnx gnrninu wyjr tlsucnireg. Vte ncetnsai, s eoumsrcn tlv isednitotan a.b.c ffjw nrv ieecrve mgeassse nzrx rk rcqr nsdtinetiao ltvm nhatero aicnetsn jn prv leucstr.

The next chapter discusses asynchronous data and event streams.

Summary

The event bus is the preferred way for verticles to communicate, and it uses asynchronous message passing.
The event bus implements both publish/subscribe (one-to-many) and point-to-point (many-to-one) communications.
While it looks like a traditional message broker, the event bus does not provide durability guarantees, so it must only be used for transient data.
Clustering allows networked instances to communicate over the distributed event bus in a transparent fashion, and to scale the workload across several application instances.

1.For a thorough discussion of using Kafka, see Dylan Scott’s Kafka in Action (Manning, 2019).

2.The W3C specification for server-sent events is available at www.w3.org/TR/eventsource.