How to convert a simple method that returns the List<String> into Multi<String> based on Smallrye Mutiny?

[Aravinda93]

I am developing an application that reads the XML file and creates the Hash ID based on the details present in XML. As of now, everything is working perfectly, and able to get the List<String>.

I would like to convert this application into Reactive Streams using the Smallrye Mutiny so I went through some of the documentation but did not understand clearly how to convert this application into Reactive Streams where I do not have to wait for the completion of all XML file to return the List<String>. Rather I can start returning the Multi<String> as and when the its generated.

Following is the simple XML that I am reading using SAX Parser to create the Hash ID:

<customerList>
    <customer>
        <name>Batman</name>
        <age>25</age>
    </customer>
    <customer>
        <name>Superman</name>
        <age>28</age>
    </customer>
</customerList>

Following is the Main application which will make call to SaxHandler:

public class HashGenerator {
    public List<String> xmlHashGenerator(final InputStream xmlStream, final String hashAlgorithm) throws SAXException, ParserConfigurationException, IOException {
        final SAXParserFactory factory = SAXParserFactory.newInstance();
        final SaxHandler saxHandler = new SaxHandler(hashAlgorithm);
        factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true);
        factory.newSAXParser().parse(xmlStream, saxHandler);
        return saxHandler.getEventHashIds();
    }
}

Following is the SaxHandler which will read the XML and create HashIDs:

public class SaxHandler extends DefaultHandler {
    @Getter
    private final List<String> eventHashIds = new ArrayList<>();
    private final HashMap<String, String> contextHeader = new HashMap<>();
    private final String hashAlgorithm;
    private ContextNode currentNode = null;
    private ContextNode rootNode = null;
    private final StringBuilder currentValue = new StringBuilder();

    public SaxHandler(final String hashAlgorithm) {
        this.hashAlgorithm = hashAlgorithm;
    }

    @Override
    public void startElement(final String uri, final String localName, final String qName, final Attributes attributes) {
        if (rootNode == null && qName.equals("customer")) {
            rootNode = new ContextNode(contextHeader);
            currentNode = rootNode;
            rootNode.children.add(new ContextNode(rootNode, "type", qName));
        }else if (currentNode != null) {
            ContextNode n = new ContextNode(currentNode, qName, (String) null);
            currentNode.children.add(n);
            currentNode = n;
        }
    }

    @Override
    public void characters(char[] ch, int start, int length) {
        currentValue.append(ch, start, length);
    }

    @Override
    public void endElement(final String uri, final String localName, final String qName) {

        if (rootNode != null && !qName.equals("customer")) {
            final String value = !currentValue.toString().trim().equals("") ? currentValue.toString().trim() : null;
            currentNode.children.add(new ContextNode(currentNode, qName, value));
        }

        if (qName.equals("customer")) {
            final String preHashString = Objects.requireNonNull(rootNode).toString();
            try {
                final StringBuilder hashId = new StringBuilder();
                byte[] digest;
                digest = MessageDigest.getInstance("SHA-224").digest(preHashString.getBytes(StandardCharsets.UTF_8));
                hashId.append(DatatypeConverter.printHexBinary(digest).toLowerCase());
                eventHashIds.add(hashId.toString());
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }
            rootNode = null;
        }
        currentValue.setLength(0);
    }
}

Following is the Test:

@Test
    public void xmlTest() throws Exception {
        final HashGenerator eventHashGenerator = new HashGenerator();
        final InputStream xmlStream = getClass().getResourceAsStream("/customer.xml");
        final List<String> eventHashIds = eventHashGenerator.xmlHashGenerator(xmlStream, "sha3-256");
        System.out.println("\nGenerated Event Hash Ids : \n" + eventHashIds);
    }

Can someone please guide me to some example or provide some idea on how to convert this application into SmallRye Mutinty Multi<String> based application?

[jponge]

Your code is using blocking I/O (InputStream / getResourceAsStream, the parsing is synchronous, etc) so there is little you can do AFAIK. You might create a Uni<String>, defer the XML parsing to a worker thread, but this is not going to be "reactive" per-se.

[Aravinda93]

@jponge Thanks a lot for the response. Just to make it Reactive, let's forget the getResourceAsStream part which is in the Junit Test case. If we remove that out of our context then how can we convert the method public List<String> xmlHashGenerator within HashGenerator to Reactive-Streams?

[jponge]

You might look at Multi.createFrom().emitter(...)

[Aravinda93]

@jponge Thanks a lot for your response. Based on your suggestion I changed my method to something like this, can you please once have a look and suggest if this is a good approach? Because even here I feel we need to wait for all the elements in the List to complete before emitting the response.

public Multi<String> xmlEventHashGenerator(final InputStream xmlStream) throws SAXException, ParserConfigurationException, IOException {
    final SAXParserFactory factory = SAXParserFactory.newInstance();
    final SaxHandler saxHandler = new SaxHandler();
    factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true);
    factory.newSAXParser().parse(xmlStream, saxHandler);

    return Multi.createFrom().emitter(em ->{
        saxHandler.getRootNodes().forEach(contextNode -> {
            final String preHashString = contextNode.toString();
            try {
                final StringBuilder hashId = new StringBuilder();
                MessageDigest.getInstance("SHA-256").digest(preHashString.getBytes(StandardCharsets.UTF_8));
                hashId.append(DatatypeConverter.printHexBinary(digest).toLowerCase());
                em.emit(hashId.toString());
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }
        });
        em.complete();
    });
}

[jponge]

Yes you basically await for all elements then wrap a list into a Multi, but it's not very "reactive".

[jponge]

What you could do is emit the SAX events in a Multi (using the emitter to do the adaptation), then have transformations (onItem().transform(...)) to do the SHA-256 computations, and either subscribe or emit the events from another thread (see https://smallrye.io/smallrye-mutiny/guides/emission-threads).

But be aware that your XML parser still relies on blocking I/O, so it won't be truly reactive anyway.

[Aravinda93]

@jponge Thanks a lot again for the response.

I am a bit unable to understand completely what you are referring to in this comment:

What you could do is emit the SAX events in a Multi (using the emitter to do the adaptation), then have transformations (onItem().transform(...)) to do the SHA-256 computations, and either subscribe or emit the events from another thread

If possible can you please provide some sort of example? Thanks a lot in advance :)

[jponge]

So I'm not into your context and I can't write your code, but here's something that should help.

It's not too hard to adapt SAX parsing events and forward them in a Multi, and dispatch the subscription to a thread pool (because the XML parsers use blocking I/O, so it's never going to be fully reactive).

Let's have:

<?xml version="1.0" encoding="UTF-8"?>
<some-document>
    <yolo>Yolo!</yolo>
    <foo bar="123">baz</foo>
</some-document>

Then we can adapt some of the XML parsing events and forward them as strings in a Multi:

///usr/bin/env jbang "$0" "$@" ; exit $?
//DEPS io.smallrye.reactive:mutiny:1.6.0

import java.io.File;
import java.io.IOException;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.parsers.SAXParserFactory;

import org.xml.sax.Attributes;
import org.xml.sax.SAXException;
import org.xml.sax.helpers.DefaultHandler;

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.infrastructure.Infrastructure;

public class main {

    public static void main(String... args) throws Throwable {
        var multi = Multi.createFrom().emitter(emitter -> {
            try {
                var parser = SAXParserFactory.newInstance().newSAXParser();
                parser.parse(new File("yolo.xml"), new DefaultHandler() {

                    @Override
                    public void startElement(String uri, String localName, String qName,Attributes attributes) throws SAXException {
                        for (int i = 0; i < attributes.getLength(); i++) {
                            emitter.emit(attributes.getQName(i) + "=" + attributes.getValue(i));
                        }
                        emitter.emit(qName);
                    }

                    @Override
                    public void endDocument() throws SAXException {
                        emitter.complete();
                    }
                });
            } catch (ParserConfigurationException | SAXException | IOException e) {
                emitter.fail(e);
            }
        });

        multi
            .runSubscriptionOn(Infrastructure.getDefaultExecutor())
            .onItem().transform(item -> "> " + item)
            .subscribe().with(System.out::println, Throwable::printStackTrace, () -> System.out.println("== Done =="));
    }
}

Of course the code is quite stupid, you might want to forward event objects instead of strings, so you can know what each item is for, you will want to deal with text content, etc, but this really shows you how to take XML parsing events and turn them into some Multistream.

$ jbang run main.java
[jbang] Building jar...
> some-document
> yolo
> bar=123
> foo
== Done ==

From here since it's a Multi you can do all the processing you want: filtering, concatenating, transforming, etc.

Hope it helps!