dataartisans / da-streamingledger Goto Github PK

Stargazers

Forkers

harshach xuqianjin-stars zzwgit wwjiang007 aiyi2099 mylinyuzhi lemonhall chinpeng adonis2014 awesomedatatool lydpolaris phan-pivotal bitsilly tryallcomingthing gavinljj drag0nglass whitespur kimmking xiashuijun john-tipper knaufk cole2295 sugar-foxs ys610zz 1261823 dineshdataengineer chenqin yx91490 moh-esmaeilzadeh huaxuehuaxue iq-scm

StreamingLedger.State<String, Int> roomAvailabilities = tradeLedger.declareState("roomAvailabilities")
                .withKeyType(String.class)
                .withValueType(Int.class);

OutputTag<BookingResult> transactionResults = tradeLedger.usingStream(bookings, "bookings")
                .apply(new CheckAndUpdateAvailabilityHandler())
                .on(roomAvailabilities, Booking::getRoomIds, "rooms", READ_WRITE)
                .output();

static final class CheckAndUpdateAvailabilityHandler extends TransactionProcessFunction<Booking, BookingResult> {

        @ProcessTransaction
        public void Booking(
                final Booking book,
                final Context<Booking> ctx,
                final @State("rooms") StateAccess<Int> rooms) {
//and In here I can access to rooms and check availability of all of them
        }
    }