i have this utility class made with  webflux:
package com.mediaworld.apimanager.utilities;

import com.mediaworld.apimanager.models.LoginRequestElement;
import jakarta.xml.soap.*;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

@Component
public class ReactiveSoapUtilities {

    public Mono<SOAPMessage> createSoapRequestReactive(Mono<LoginRequestElement> loginRequestElement){

        return loginRequestElement.flatMap( requestElement -> {
            SOAPMessage request = null;

            try {
                MessageFactory messageFactory = MessageFactory.newInstance(SOAPConstants.SOAP_1_1_PROTOCOL);
                request = messageFactory.createMessage();

                SOAPPart messagePart = request.getSOAPPart();
                SOAPEnvelope envelope = messagePart.getEnvelope();

                //envelope.removeNamespaceDeclaration(envelope.getPrefix());
                envelope.setPrefix("SOAP-ENV");
                /*SOAPHeader header = request.getSOAPHeader();
                header.setPrefix("soap-env-3");*/


                /*String soapAction = "/LoginService/login";
                request.getMimeHeaders().addHeader("SOAPAction",soapAction);*/

                SOAPBody body = envelope.getBody();
                body.setPrefix("SOAP-ENV");

                SOAPElement loginRequestElement1 = body.addChildElement("loginRequestElement","ns0","http://www.tibco.com/schemas/login_bw_project/login/schemas/Schema.xsd");

                SOAPElement userElement = loginRequestElement1.addChildElement("user","ns0");
                userElement.setTextContent(requestElement.getUser());

                SOAPElement passwordElement = loginRequestElement1.addChildElement("pass","ns0");
                passwordElement.setTextContent(requestElement.getPassword());

                request.writeTo(System.out);

                return Mono.just(request);
            }catch (Exception e){
                return Mono.error(e);
            }

        });
    }
}


then i have this service class which calls it, but i need to modify it because it is not implemented to read a Mono<SOAPMessage>, how can i modify in order to make it work properly?

package com.mediaworld.apimanager.testActiveMQ.service;

import com.mediaworld.apimanager.models.LoginRequestElement;
import com.mediaworld.apimanager.utilities.ReactiveSoapUtilities;
import jakarta.jms.TextMessage;
import jakarta.xml.soap.SOAPException;
import jakarta.xml.soap.SOAPMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

@Service
public class ReactiveProducerService {

    @Value("${spring.activemq.topic}")
    String topic;

    @Value("${spring.activemq.queue}")
    String queue;

    @Autowired
    JmsTemplate jmsTemplate;

    @Autowired
    private ReactiveSoapUtilities reactiveSoapUtilities;

    public void sendToQueue(LoginRequestElement loginRequestElement){
        try {
            // employee = new Employee("Micheal", "Jackson", 10000L, new Date(), 24);
            //String jsonObj = new XmlMapper().writer().withDefaultPrettyPrinter().writeValueAsString(loginRequestElement);
            jmsTemplate.send(queue, messageCreator -> {
                TextMessage message = messageCreator.createTextMessage();
                //message.setText(jsonObj);
                SOAPMessage message1 = reactiveSoapUtilities.createSoapRequest(loginRequestElement);
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                try {
                    message1.writeTo(out);
                } catch (SOAPException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                String strMsg = new String(out.toByteArray());
                message.setText(strMsg);
                return message;
            });

        }
        catch (Exception ex) {
            System.out.println("ERROR in sending message to queue");
        }
    }
}