public Flux < ServiceMessage > requestBidirectional ( Publisher < ServiceMessage > publisher , Class < ? > responseType , Address address ) { return Flux . defer ( ( ) -> { requireNonNull ( address , "requestBidirectional address parameter is required and must not be null" ) ; requireNonNull ( transport , "transport is required and must not be null" ) ; return transport . create ( address ) . requestChannel ( publisher ) . map ( message -> ServiceMessageCodec . decodeData ( message , responseType ) ) . map ( this :: throwIfError ) ; } ) ; }
Given an address issues stream of service requests to service which returns stream of service messages back .