RSocketRequester

RSocketRequester RSocket so'rovlarini bajarish uchun ochiq API taqdim etadi va bu kam darajadagi ma'lumotlar buferlari o'rniga ma'lumotlar va metama'lumotlar uchun ob'ektlarni qabul qiladi va qaytaradi. U simmetrik ravishda, mijozlar tomonidan so'rovlarni bajarish uchun va serverlar tomonidan so'rovlarni bajarish uchun ishlatilishi mumkin.

Mijoz tomonda requester

Mijoz tomonda RSocketRequester olish uchun siz serverga ulanish kerak, bu esa RSocket protokolining SETUP freymasini ulanish sozlamalari bilan yuborishni talab qiladi. RSocketRequester sizga io.rsocket.core.RSocketConnector ni sozlashda yordam beradigan qurish vositasini taqdim etadi, shu jumladan SETUP freymasi uchun ulanish sozlamalarini.

Bu default sozlamalar bilan ulanishning eng oddiy usuli:

RSocketRequester requester = RSocketRequester.builder().tcp("localhost", 7000);
URI url = URI.create("https://example.org:8080/rsocket");
RSocketRequester requester = RSocketRequester.builder().webSocket(url);

val requester = RSocketRequester.builder().tcp("localhost", 7000)
URI url = URI.create("https://example.org:8080/rsocket");
val requester = RSocketRequester.builder().webSocket(url)

Yuqoridagi holatda ulanish darhol o'rnatilmaydi. So'rovlar kelib tushganda ulanish avtomatik ravishda o'rnatiladi va ishlatiladi.

Ulanishni sozlash

RSocketRequester.Builder SETUP freymasining dastlabki sozlamalarini quyidagi kabi qilib sozlash uchun imkoniyat beradi:

• dataMimeType(MimeType) – ulanish uchun ma'lumotlar MIME tipini o'rnatadi.

• metadataMimeType(MimeType) – ulanish uchun metama'lumotlar MIME tipini o'rnatadi.

• setupData(Object)– SETUP ga kiritiladigan ma'lumotlar.

• setupRoute(String, Object…) – SETUP ga kiritiladigan metama'lumotlardagi marshrut.

• setupMetadata(Object, MimeType)– SETUP ga kiritiladigan boshqa metama'lumotlar.

Ma'lumotlar uchun odatiy MIME tipi birinchi bo'lib sozlangan Decoder bo'yicha belgilanadi. Metama'lumotlar uchun odatiy MIME tipi – kompozit metama'lumotlar, bu har bir so'rov uchun bir nechta metama'lumotlar va MIME tiplarini ishlatishga imkon beradi. Odatda, ikkalasini ham almashtirish talab etilmaydi.

SETUP freymidagi ma'lumotlar va metama'lumotlar ixtiyoriy hisoblanadi. Server tomonida @ConnectMapping annotatsiyasi bilan belgilangan metodlar ulanish bosqichini va SETUP freymidagi tarkibni boshqarish uchun ishlatilishi mumkin. Metama'lumotlar ulanish darajasida xavfsizlikni ta'minlash uchun ishlatilishi mumkin.

Strategiyalar

RSocketRequester.Builder requester ni konfiguratsiya qilish uchun RSocketStrategies qabul qiladi. Siz uni freymalar va ma'lumotlar va metama'lumotlar qiymatlarini (de)-seriyalizatsiya qilish uchun dekoderlar dan foydalanish kerak bo'ladi. Default bo'yicha spring-core dan faqat String, byte[] va ByteBuffer ga asosiy codec lar ro'yxatdan o'tgan. spring-web ni qo'shish quyidagi tarzda ko'proq imkoniyatlar ochib beradi:

RSocketStrategies strategies = RSocketStrategies.builder()
.encoders(encoders -> encoders.add(new Jackson2CborEncoder()))
.decoders(decoders -> decoders.add(new Jackson2CborDecoder()))
.build();
RSocketRequester requester = RSocketRequester.builder()
.rsocketStrategies(strategies)
.tcp("localhost", 7000);

val strategies = RSocketStrategies.builder()
.encoders { it.add(Jackson2CborEncoder()) }
.decoders { it.add(Jackson2CborDecoder()) }
.build()
val requester = RSocketRequester.builder()
.rsocketStrategies(strategies)
.tcp("localhost", 7000)

RSocketStrategies qayta foydalanish uchun mo'ljallangan. Ba'zi hollarda, masalan, mijoz va server bir dasturda bo'lsa, bu Spring konfiguratsiyasida e'lon qilish afzalroq bo'lishi mumkin.

Mijoz tomonda responder lar

RSocketRequester.Builder serverdan keladigan so'rovlarga javob beradigan responder larni sozlash uchun foydalanilishi mumkin.

Server qismini ta'minlaydigan infratuzilma asosida, lekin quyidagicha dasturiy yo'l bilan ro'yxatdan o'tish mumkin bo'lgan annotatsiyalangan handler lar yordamida mijoz tomonda javob berish mumkin:

RSocketStrategies strategies = RSocketStrategies.builder()
.routeMatcher(new PathPatternRouteMatcher())  
.build();
SocketAcceptor responder =
RSocketMessageHandler.responder(strategies, new ClientHandler()); 
RSocketRequester requester = RSocketRequester.builder()
.rsocketConnector(connector -> connector.acceptor(responder)) 
.tcp("localhost", 7000);

val strategies = RSocketStrategies.builder()
.routeMatcher(PathPatternRouteMatcher())  
.build()
val responder =
RSocketMessageHandler.responder(strategies, new ClientHandler()); 
val requester = RSocketRequester.builder()
.rsocketConnector { it.acceptor(responder) } 
.tcp("localhost", 7000)

Yuqoridagi qisqartma faqat mijoz responder lari uchun dasturiy ro'yxatdan o'tkazish uchun mo'ljallangan. Alternativ ssenariylarda, agar mijoz responder lari Spring konfiguratsiyasida joylashgan bo'lsa, RSocketMessageHandler ni Spring bean sifatida e'lon qilish va keyinchalik quyidagicha foydalanish mumkin:

ApplicationContext context = ... ;
RSocketMessageHandler handler = context.getBean(RSocketMessageHandler.class);
RSocketRequester requester = RSocketRequester.builder()
.rsocketConnector(connector -> connector.acceptor(handler.responder()))
.tcp("localhost", 7000);

import org.springframework.beans.factory.getBean
val context: ApplicationContext = ...
val handler = context.getBean<RSocketMessageHandler>()
val requester = RSocketRequester.builder()
.rsocketConnector { it.acceptor(handler.responder()) }
.tcp("localhost", 7000)

Yuqorida tasvirlangan narsa ham RSocketMessageHandler da setHandlerPredicate dan foydalanishni talab qilishi mumkin, bu esa mijoz responder larini aniqlashning boshqa strategiyasiga o'tish uchun, masalan, @RSocketClientResponder kabi foydalanuvchi annotatsiyasi asosida, standart @Controller emas. Bu mijoz va server yoki bir xil dasturda bir nechta mijozlar mavjud bo'lgan ssenariylarda kerak.

Kengaytirilgan rejim

RSocketRequesterBuilder asosiy io.rsocket.core.RSocketConnector ni qo'shimcha konfiguratsiya parametrlariga, masalan, "keepalive" interval parametrlari, seansni tiklash, "interceptors" va h.k. kirish uchun ochuvchi callback ni amalga oshirish imkonini beradi. Bu darajada parametrlarni quyidagi tarzda sozlashingiz mumkin:

RSocketRequester requester = RSocketRequester.builder()
.rsocketConnector(connector -> {
// ...
})
.tcp("localhost", 7000);

val requester = RSocketRequester.builder()
.rsocketConnector {
    //...
}
.tcp("localhost", 7000)

Server tomonining requester

Serverdan ulanishgan mijozlarga so'rovlarni yuborish uchun, serverdan ulanishgan mijoz uchun requester olish kerak bo'ladi.

Annotatsiyalangan responder larda @ConnectMapping va @MessageMapping annotatsiyalariga ega bo'lgan metodlar RSocketRequester argumentini qo'llab-quvvatlaydi. Uni ulanish uchun konkret requester ga kirish imkoniyatini olish uchun ishlatishingiz mumkin. Eslab qoling, @ConnectMapping annotatsiyasi bilan belgilangan metodlar aslida SETUP freymasini qayta ishlovchi hisoblanadi, bu so'rovlar yuborishdan oldin qayta ishlanishi lozim. Shuning uchun birinchi bosqichdagi so'rovlarni qayta ishlash jarayonidan ajratish kerak. Misol:

@ConnectMapping
Mono<Void> handle(RSocketRequester requester) {
requester.route("status").data("5")
.retrieveFlux(StatusReport.class)
.subscribe(bar -> { 
    // ...
});
return ... 
}

@ConnectMapping
suspend fun handle(requester: RSocketRequester) {
GlobalScope.launch {
requester.route("status").data("5").retrieveFlow<StatusReport>().collect { 
    // ...
}
}
/// ... (2)
}

So'rovlar

Mijoz yoki server uchun requester olingandan so'ng, so'rovlarni quyidagicha yuborishingiz mumkin:

ViewBox viewBox = ... ;
Flux<AirportLocation> locations = requester.route("locate.radars.within") 
.data(viewBox) 
.retrieveFlux(AirportLocation.class); 

val viewBox: ViewBox = ...
val locations = requester.route("locate.radars.within") 
.data(viewBox) 
.retrieveFlow<AirportLocation>() 

O'zaro ta'sir turi kiritish va chiqarish kartinaligining (cardinality) kuchi orqali bilvosita belgilanadi. Yuqoridagi misol Request-Stream, chunki bir taqiqlik qiymat yuborilib, qiymatlar oqimi qaytariladi. Bunda, agar RSocket o'zaro ta'sir turi va responder kutilyotgan kirish va chiqish turlariga mos kelayotgan bo'lsa, ko'p o'ylash kerak emas. Yagona noto'g'ri kombinatsiya bu "ko'pdan birga" kombinatsiyasidir.

data(Object) metodi shuningdek Reactive Streams spetsifikatsiyasiga kiruvchi har qanday Publisher ni qabul qiladi, shu jumladan Flux va Mono, shuningdek ReactiveAdapterRegistry da ro'yxatdan o'tgan har qanday qiymat generatorlari. Agar Flux kabi bir xil turdagi qiymatlar kiritadigan ko'p qiymatli Publisher mavjud bo'lsa, har bir element uchun turini tekshirish va Encoder qidirishni bartaraf etish uchun data ning biror bir o'rnatilgan metodidan foydalanishni o'ylab ko'ring:

data(Object producer, Class<?> elementClass);
data(Object producer, ParameterizedTypeReference<?> elementTypeRef);

data(Object) bosqichi ixtiyoriydir. Ma'lumot yubormaydigan so'rovlar holatida uni o'tkazib yuboramiz:

Mono<AirportLocation> location = requester.route("find.radar.EWR"))
.retrieveMono(AirportLocation.class);

import org.springframework.messaging.rsocket.retrieveAndAwait
val location = requester.route("find.radar.EWR")
.retrieveAndAwait<AirportLocation>()

Agar kompozit metama'lumotlar (default holatda) ishlatilsa va qiymatlar qo'llab-quvvatlanayotgan Encoder bilan ro'yxatdan o'tgan bo'lsa, qo'shimcha metama'lumotlarni qo'shish mumkin. Misol:

String securityToken = ... ;
ViewBox viewBox = ... ;
MimeType mimeType = MimeType.valueOf("message/x.rsocket.authentication.bearer.v0");
Flux<AirportLocation> locations = requester.route("locate.radars.within")
.metadata(securityToken, mimeType)
.data(viewBox)
.retrieveFlux(AirportLocation.class);

import org.springframework.messaging.rsocket.retrieveFlow
val requester: RSocketRequester = ...
val securityToken: String = ...
val viewBox: ViewBox = ...
val mimeType = MimeType.valueOf("message/x.rsocket.authentication.bearer.v0")
val locations = requester.route("locate.radars.within")
.metadata(securityToken, mimeType)
.data(viewBox)
.retrieveFlow<AirportLocation>()

Fire-and-Forget uchun send() metodidan foydalanish kerak, bu Mono<Void> ni qaytaradi. E'tibor bering, Mono faqatgina xabar muvaffaqiyatli yuborilganini, lekin u qayta ishlanganini ko'rsatmaydi.

Metadata-Push uchun sendMetadata() metodidan foydalaning, u Mono<Void> ni qaytaradi.