Аннотированные респондеры

RSocket-респондеры могут быть реализованы в виде методов с аннотациями @MessageMapping и @ConnectMapping. Методы, аннотированные @MessageMapping, обрабатывают отдельные запросы, а методы, аннотированные @ConnectMapping, – события на уровне соединения (установка и проталкивание метаданных). Аннотированные респондеры поддерживаются симметрично, и для ответов со стороны сервера, и для ответов со стороны клиента.

Респондеры на стороне сервера

Чтобы использовать аннотированные респондеры на стороне сервера, добавьте RSocketMessageHandler в конфигурацию Spring для обнаружения помеченных аннотацией @Controller бинов с методами, аннотированными @MessageMapping и @ConnectMapping:

@Configuration
static class ServerConfig {
@Bean
public RSocketMessageHandler rsocketMessageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.routeMatcher(new PathPatternRouteMatcher());
return handler;
}
}

@Configuration
class ServerConfig {
@Bean
fun rsocketMessageHandler() = RSocketMessageHandler().apply {
routeMatcher = PathPatternRouteMatcher()
}
}

Затем запустите RSocket-сервер через API RSocket для Java и подключите RSocketMessageHandler для респондера следующим образом:

ApplicationContext context = ... ;
RSocketMessageHandler handler = context.getBean(RSocketMessageHandler.class);
CloseableChannel server =
RSocketServer.create(handler.responder())
.bind(TcpServerTransport.create("localhost", 7000))
.block();

import org.springframework.beans.factory.getBean
val context: ApplicationContext = ...
val handler = context.getBean<RSocketMessageHandler>()
val server = RSocketServer.create(handler.responder())
.bind(TcpServerTransport.create("localhost", 7000))
.awaitSingle()

RSocketMessageHandler по умолчанию поддерживает составные метаданные и метаданные маршрутизации. Можно настроить его MetadataExtractor, если требуется переключиться на другой MIME-тип или зарегистрировать дополнительные MIME-типы метаданных.

Необходимо будет настроить экземпляры Encoder и Decoder, которые требуются для поддержки метаданных и форматов данных. Для реализации кодеков, скорее всего, понадобится модуль spring-web.

По умолчанию SimpleRouteMatcher используется для согласования маршрутов через AntPathMatcher. Рекомендуем подключить PathPatternRouteMatcher из spring-web для эффективного согласования маршрутов. Маршруты RSocket могут быть иерархическими, но не являются URL-путями. Оба средства сопоставления маршрутов настроены на использование "." в качестве разделителя по умолчанию, а декодирование URL-адресов, как в случае с URL-адресами HTTP, отсутствует.

RSocketMessageHandler можно сконфигурировать через RSocketStrategies, что может быть полезно, если необходимо разделить конфигурацию между клиентом и сервером в одном и том же процессе:

@Configuration
static class ServerConfig {
@Bean
public RSocketMessageHandler rsocketMessageHandler() {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.setRSocketStrategies(rsocketStrategies());
return handler;
}
@Bean
public RSocketStrategies rsocketStrategies() {
return RSocketStrategies.builder()
    .encoders(encoders -> encoders.add(new Jackson2CborEncoder()))
    .decoders(decoders -> decoders.add(new Jackson2CborDecoder()))
    .routeMatcher(new PathPatternRouteMatcher())
    .build();
}
}

@Configuration
class ServerConfig {
@Bean
fun rsocketMessageHandler() = RSocketMessageHandler().apply {
rSocketStrategies = rsocketStrategies()
}
@Bean
fun rsocketStrategies() = RSocketStrategies.builder()
    .encoders { it.add(Jackson2CborEncoder()) }
    .decoders { it.add(Jackson2CborDecoder()) }
    .routeMatcher(PathPatternRouteMatcher())
    .build()
}

Респондеры на стороне клиента

Аннотированные респондеры на стороне клиента должны быть сконфигурированы в RSocketRequester.Builder.

@MessageMapping

После того, как конфигурация респондера для server или client будет настроена, методы с аннотацией @MessageMapping можно будет использовать следующим образом:

@Controller
public class RadarsController {
@MessageMapping("locate.radars.within")
public Flux<AirportLocation> radars(MapRequest request) {
// ...
}
}

@Controller
class RadarsController {
@MessageMapping("locate.radars.within")
fun radars(request: MapRequest): Flow<AirportLocation> {
// ...
}
}

Приведенный выше метод, помеченные аннотацией @MessageMapping, реагирует на взаимодействие типа Request-Stream с маршрутом "locate.radars.within". Он поддерживает гибкую сигнатуру метода с возможностью использования следующих аргументов метода:

Ожидается, что возвращаемым значением будет один или несколько объектов, которые будут сериализованы в качестве полезных данных ответа. Это могут быть асинхронные типы, такие как Mono или Flux, конкретное значение, либо void или асинхронный тип без значения, такой как Mono<Void>.

Тип взаимодействия RSocket, который поддерживает метод, аннотированный @MessageMapping, определяется по мощности множества ввода (т.е. аргумента, аннотированного @Payload) и вывода, где мощность множества означает следующее:

В таблице ниже представлены все комбинации мощности множества ввода и вывода и соответствующий тип(ы) взаимодействия:

@ConnectMapping

Аннотация @ConnectMapping обрабатывает фрейм SETUP в начале установления RSocket-соединения и все последующие уведомления о проталкивании метаданных через фрейм METADATA_PUSH, т.е. metadataPush(Payload) в io.rsocket.RSocket.

Методы с аннотацией @ConnectMapping поддерживают те же аргументы, что и методы с аннотацией @MessageMapping, но основаны на метаданных и данных из фреймов SETUP и METADATA_PUSH. Аннотация @ConnectMapping может быть снабжена шаблоном для отграничения процедуры обработки определенными соединениями, которые имеют маршрут в метаданных, а если шаблоны не будут объявлены, то все соединения будут соответствовать.

Методы, аннотированные @ConnectMapping, не могут возвращать данные, поэтому их нужно объявлять с использованием void или Mono<Void> в качестве возвращаемого значения. Если процедура обработки возвращает ошибку для нового соединения, то соединение отклоняется. Процедура обработки не должна затягиваться, чтобы можно было выполнить запросы к RSocketRequester для конкретного соединения.