javaspringreactive-programmingspring-webflux

Hot to get body as String from spring reactive ClientRequest?


In a test method an instance of org.springframework.web.reactive.function.client.ClientRequest is received.

I want to verify its HttpMethod, URI and body.

It is quite obvious how to get everything except for the body.

ClientRequest request = makeInstance(...);

assertEquals(HttpMethod.POST, request.method());
assertEquals("somewhere/else", request.url().toString());

// ? unclear how to extract body using the BodyInserter

BodyInserter<?, ? super ClientHttpRequest> inserter = request.body();

inserter.insert(%outputMessage%, %context%);

I have found in sources of Spring how BodyInserters are tested. It is more or less clear how to create a BodyInserter.Context (the second parameter), but I cannot understand how to construct the first parameter, so the request body could be extracted through it.

Please, show a conventional (or at least usable) way to get request body from a ClientRequest instance.


Solution

  • A bit complex for such a simple case, but it took me to implement 5 classes, to extract a body from ClientRequest.

    It seems to be too much, and I'm still curious if there is a short solution for the problem. You're welcome to propose another answer to this question, so I could accept it.

    Unfortunately, it must be said that the design of ClientRequest, BodyInserters and most of the other stuff from org.springframework.web.reactive.*** has a large room for improvement. At the moment it is just a pile of interfaces with tons of methods per each, and it usually takes a lot of efforts to test code, depending on those classes.

    The main goal it to get this method working:

    static <T> T extractBody(ClientRequest request, Class<T> clazz) {
    
      InsertionReceiver<T> receiver = InsertionReceiver.forClass(clazz);
      return receiver.receiveValue(request.body());
    }
    

    Here is the implementation of InsertionReceiver:


    import org.springframework.http.ReactiveHttpOutputMessage;
    import org.springframework.web.reactive.function.BodyInserter;
    
    public interface InsertionReceiver<T> {
    
      T receiveValue(BodyInserter<?, ? extends ReactiveHttpOutputMessage> bodyInserter);
    
      static <T> InsertionReceiver<T> forClass(Class<T> clazz) {
        return new SimpleValueReceiver<>(clazz);
      }
    }
    

    import java.util.concurrent.atomic.AtomicReference;
    import org.springframework.http.ReactiveHttpOutputMessage;
    import org.springframework.web.reactive.function.BodyInserter;
        
    class SimpleValueReceiver<T> implements InsertionReceiver<T> {
    
      private static final Object DUMMY = new Object();
    
      private final Class<T> clazz;
      private final AtomicReference<Object> reference;
    
      SimpleValueReceiver(Class<T> clazz) {
        this.clazz = clazz;
        this.reference = new AtomicReference<>(DUMMY);
      }
    
      @Override
      public T receiveValue(BodyInserter<?, ? extends ReactiveHttpOutputMessage> bodyInserter) {
        demandValueFrom(bodyInserter);
    
        return receivedValue();
      }
    
      private void demandValueFrom(BodyInserter<?, ? extends ReactiveHttpOutputMessage> bodyInserter) {    
        var inserter = (BodyInserter<?, ReactiveHttpOutputMessage>) bodyInserter;
    
        inserter.insert(
            MinimalHttpOutputMessage.INSTANCE,
            new SingleWriterContext(new WriteToConsumer<>(reference::set))
        );
      }
    
      private T receivedValue() {
        Object value = reference.get();
        reference.set(DUMMY);
    
        T validatedValue;
    
        if (value == DUMMY) {
          throw new RuntimeException("Value was not received, Check your inserter worked properly");
        } else if (!clazz.isAssignableFrom(value.getClass())) {
          throw new RuntimeException(
              "Value has unexpected type ("
                  + value.getClass().getTypeName()
                  + ") instead of (" + clazz.getTypeName() + ")");
        } else {
          validatedValue = clazz.cast(value);
        }
    
        return validatedValue;
      }
    }
    

    class WriteToConsumer<T> implements HttpMessageWriter<T> {
    
      private final Consumer<T> consumer;
      private final List<MediaType> mediaTypes;
    
      WriteToConsumer(Consumer<T> consumer) {
        this.consumer = consumer;
        this.mediaTypes = Collections.singletonList(MediaType.ALL);
      }
    
      @Override
      public List<MediaType> getWritableMediaTypes() {
        return mediaTypes;
      }
    
      @Override
      public boolean canWrite(ResolvableType elementType, MediaType mediaType) {
        return true;
      }
    
      @Override
      public Mono<Void> write(
          Publisher<? extends T> inputStream,
          ResolvableType elementType,
          MediaType mediaType,
          ReactiveHttpOutputMessage message,
          Map<String, Object> hints
      ) {
        inputStream.subscribe(new OneValueConsumption<>(consumer));
        return Mono.empty();
      }
    }
    

    class MinimalHttpOutputMessage implements ReactiveHttpOutputMessage {
    
      public static MinimalHttpOutputMessage INSTANCE = new MinimalHttpOutputMessage();
    
      private MinimalHttpOutputMessage() {
      }
    
      @Override
      public HttpHeaders getHeaders() {
        return HttpHeaders.EMPTY;
      }
    
      // other overridden methods are omitted as they do nothing,
      // i.e. return null, false, or have empty bodies
    }
    

    class OneValueConsumption<T> implements Subscriber<T> {
    
      private final Consumer<T> consumer;
      private int remainedAccepts;
    
      public OneValueConsumption(Consumer<T> consumer) {
        this.consumer = Objects.requireNonNull(consumer);
        this.remainedAccepts = 1;
      }
    
      @Override
      public void onSubscribe(Subscription s) {
        s.request(1);
      }
    
      @Override
      public void onNext(T o) {
        if (remainedAccepts > 0) {
          consumer.accept(o);
          remainedAccepts -= 1;
        } else {
          throw new RuntimeException("No more values can be consumed");
        }
      }
    
      @Override
      public void onError(Throwable t) {
        throw new RuntimeException("Single value was not consumed", t);
      }
    
      @Override
      public void onComplete() {
        // nothing
      }
    }
    

    class SingleWriterContext implements BodyInserter.Context {
    
      private final List<HttpMessageWriter<?>> singleWriterList;
    
      SingleWriterContext(HttpMessageWriter<?> writer) {
        this.singleWriterList = List.of(writer);
      }
    
      @Override
      public List<HttpMessageWriter<?>> messageWriters() {
        return singleWriterList;
      }
    
      @Override
      public Optional<ServerHttpRequest> serverRequest() {
        return Optional.empty();
      }
    
      @Override
      public Map<String, Object> hints() {
        return null;
      }
    }