Reaktiiviset laajennukset (Reactive Extensions, rx) ovat kirjastoperhe asynkroniseen ohjelmointiin. Siitä on kirjastototeutuksia käytännössä kaikille kielille ja alustoille, esimerkiksi JavaScript, Java, C#, C++ ja Swift. Rx:n voi toteuttaa mille tahansa kielelle, mutta sen helppo käyttö edellyttää kieleltä tukea lambdoille ja korkeamman kertaluvun funktioille. Tässä käsitellään ohjelmointikieliagnostisesti rx:n käsitteitä ja toimintaa.

Rx laajentaa tarkkailijasuunnittelumallin yksittäisistä toisistaan irrallisista tapahtumista tapahtumasekvenssiksi. Tämä sekvenssi voi olla loogisesti loputon, keskeytyä virheeseen tai päättyä. Oleellinen osa rx:ää ovat operaatiot näiden sekvenssien käsittelyyn deklaratiivisesti.

Kuten tarkkailijasuunnittelumallissa, rx:n oleellisimmat käsitteet ovat tarkkailija (Observer) ja tarkkailtava (Observable). Rx:n tarkkailijassa on kaksi lisäystä suunnittelumalliin: ensinnäkin se saa tiedon, kun sekvenssi päättyy, ja toiseksi sille ilmoitetaan myös virheestä. Suurin osa rx:n toiminnoista liittyy kuitenkin tarkkailtavaan. Käsitteellisenä erona tyypilliseen suunnittelumallin mukaiseen tarkkailtavaan rx:ssä tarkkailtava kuvaa yleensä yhtä ominaisuutta siinä missä suunnittelumalli kuvaa sen luokkana, jolla on useita muuttuvia ominaisuuksia. Samanlainen luokka koostuisi siis rx:ssä useammasta erikseen tarkkailtavasta ominaisuudesta. Tarkkailtavan ei kuitenkaan tarvitse olla luokan ominaisuus, vaan se voi olla myös irrallinen olio.

Tarkkailtava on rx:ssä kirjastototeutuksen tarjoama luokka, joka yleensä kytketään rx:n käyttäjän omaan tapahtumalähteeseen. Tarkkailtava työntää (push) alkionsa tarkkailijalle sekvenssissä, joka voi päättyä virheeseen tai sekvenssin päättymiseen.

Rx:n tarkkailija on rajapinta, jonka rx:n käyttäjä toteuttaa käsitelläkseen tarkkailtavan työntämiä alkioita ja käsitelläkseen virhetilanteet ja sekvenssin päättymisen.

Koska rx:n tarkkailtava on sekvenssi, on vain luontevaa, että sitä voi käsitellä kuten standardikirjaston listaa. Koska se kuitenkin on luonteeltaan asynkroninen, ei sen käsittely imperatiivisesti silmukassa ole mahdollista ilman odottamista. Sen sijaan korkeamman kertaluvun listaoperaatiot (kuten map ja filter) sopivat myös tarkkailtavien käsittelyyn, koska niille parametrina annetut funktiot voidaan ajaa myös asynkronisesti, kun tarkkailtava lähettää uuden arvon.

Esimerkiksi seuraava pseudokoodipätkä toimii sekä standardikirjaston listoille että rx:n tarkkailtavalle. Siinä missä tavalliselle listalle se suoritetaan synkronisesti joko saman tien tai laiskasti vasta, kun arvoa käytetään johonkin, tarkkailtavalle tämä suoritetaan asynkronisesti, kun uusi arvo saapuu.

fetchData()
    .map(s -> s.toInt)
    .filter(i -> i > 1000)
    .map(i -> "Fetched value: " + i)

Tässä on myös listan ja tarkkailtavan käsitteellinen ero. Listalta haetaan arvoja (pull), tarkkailtava työntää arvoja ulos (push). Jotta tarkkailtavan arvon saa ulos, on tarkkailtavalle rekisteröitävä tarkkailija. Tämä taas vastaa listan käsittelemistä forEach-funktiolla.

Tulostettaessa listaa forEach-funktiolla jokainen listan alkio tulostetaan peräkkäin, jonka jälkeen operaatio palaa. Seuraava pseudokoodipätkä esittää tätä ja se siis ajetaan kerralla alusta loppuun.

fetchData()
    .map(s -> s.toInt)
    .filter(i -> i > 1000)
    .map(i -> "Fetched value: " + i)
    .forEach(i -> println(i))

Tulostettaessa tarkkailtavan alkioita rekisteröimällä (subscribe) tarkkailtavalle kuuntelijafunktiolla, listan alkiot tulostetaan, kun ne tulevat. Operaatio palaa kuitenkin heti rekisteröinnin jälkeen. Seuraava pseudokoodipätkä esittää tämän tapauksen ja jää siten ajoon asynkronisesti koodin ajamisen jälkeen.

fetchData()
    .map(s -> s.toInt)
    .filter(i -> i > 1000)
    .map(i -> "Fetched value: " + i)
    .subscribe(i -> println(i))

Rx:llä on kaksi erityistä käyttökohdetta: muuttuvan datan esittäminen käyttöliittymässä ja datavirtojen rinnakkainen käsittely. Yleensä ensimmäiseen liittyy myös paljon jälkimmäistä. Koska rx:n tarkkailtavalla on harmonisoitu rajapinta, rx:n on mahdollista tarjota lukuisia operaatioita asynkroniseen datan käsittelyyn. Operaatiot datavirtojen yhdistämiseen, ikkunointiin, puskurointiin ja kuristamiseen ovat, tavallisten listaoperaatioiden lisäksi, kaikki tarjolla tarkkailtaville.

Nykyään on yleisesti käytössä kaksi tekniikkaa, joilla rx:n tarjoamaa toiminnallisuutta tehdään: futuurit ja takaisinkutsut. Rx:n voi ajatella yhdistävän näitä tekniikoita, sillä se on kuten lista futuureja ja sen operaatiot toimivat takaisinkutsuilla. Kaikki mitä tarkkailtavilla on mahdollista tehdä, on mahdollista myös näillä muilla tekniikoilla. Rx:ssä monet tarvittavat toiminnot ovat kuitenkin jo valmiina. Omasta koodistakin tulee selkeämpää, koska toiminnan voi suorittaa selkeissä vaiheissa. Näin myös koodi on helpompi saada toimimaan oikein.

Tämä oli lyhyt johdanto rx:ään, miksi ja mihin sitä voi käyttää. Enemmän tietoa aiheesta löytyy ReactiveX:n sivuilta tai Principles of Reactive Programming -verkkokurssilta. Myös tälle kirjoitukselle on luvassa jatkoa, ainakin tarkemman RxJava-esittelyn ja esimerkin muodossa.