Create a Transformer

1. Overview
Transformers transforms the previous xMessage from the user to one that needs to be sent next. It is essentially a microservice that based on the previous user action, returns a new xMessage that will then be shown to the user. This also enables conversion from one type
All inbound messages pass through a transformer. If a transformer is not assigned, then a null transformer is assigned to the xMessage. Since the current implementation of MessageRosa is only in Java, currently there is a limitation on the number of languages you can build the transformer on which right now is just Java.
Simply put a transformer looks something like this 
1.1 Responsibilities
Acts as a state machine and converts messages from one state to another.
Used to transform messages
Applies conversational logic (constraints or input format)
2. Creating your own Transformer
All transformers are named as <Action>Transformer; for example PDFTransformer will generate a PDF from the previous XMessage and convert it to an XMessage. Adapters should extend TransformerProvider. Thus, it needs to implement the following methods:
  public XMessage transform(XMessage nextMsg) //Takes in XMessage and return the next message.
3. Sample Code
Sample code to transform text to lowercase characters before sending to outbound
    public class DemoTransformer extends TransformerProvider {
        private final Flux<ReceiverRecord<String, String>> reactiveKafkaReceiver;
        
        @Value("${processOutbound}")
        private String processOutboundTopic;
​
        @EventListener(ApplicationStartedEvent.class)
        public void onMessage() {
            reactiveKafkaReceiver
                    .doOnNext(new Consumer<ReceiverRecord<String, String>>() {
                        @Override
                        public void accept(ReceiverRecord<String, String> stringMessage) {
                            try {
                                XMessage msg = XMessageParser.parse(new ByteArrayInputStream(stringMessage.value().getBytes()));
                                transform(msg)
                                        .subscribe(new Consumer<XMessage>() {
                                            @Override
                                            public void accept(XMessage transformedMessage) {
                                                if (transformedMessage != null) {
                                                    try {
                                                        kafkaProducer.send(processOutboundTopic, transformedMessage.toXML());
                                                    } catch (JAXBException e) {
                                                        e.printStackTrace();
                                                    }
                                                }
                                            }
                                        });
                            } catch (JAXBException e) {
                                e.printStackTrace();
                            } catch (NullPointerException e) {
                                log.error("An error occured : "+e.getMessage() + " at line no : "+ e.getStackTrace()[0].getLineNumber()
                                        +" in class : "+e.getStackTrace()[0].getClassName());
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    })
                    .doOnError(new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable e) {
                            System.out.println(e.getMessage());
                            log.error("KafkaFlux exception", e);
                        }
                    }).subscribe();
​
        }
​
        /* Transform text to lower case */
        public XMessage transform(XMessage nextMsg) {
            Payload payload = nextMsg.getPayload();
            String text = payload.getText();
            payload.setText(text.toLowerCase());
            nextMsg.setPayload(payload);
            return nextMsg;
        }
    }
All transformers with the implementation same as above exmaple will be valid. A detailed example transformer can be found here.
4. Properties of a transformer
Transformers could call external services and wait for resposes from them to process message. These transformers are called call-only transformer. An example of this transformer is the PDF service transformer which calls the PDF service to check the status of queue of previous message and responds with a response containing PDF for the user.
All transformers need to be registered. All unregistered transformers will not be acted upon. This is what essentially adds a topic to broker on which the messages are pushed. Also this requires some basic config of the max time a transformer could take to process the message.
Scaling of transformers is done horizontally but the broker needs to know the number so tha partitions can be reconfigured.
Since it's part of a state machine. If the transformer is stuck it needs communicate to the Orchestrator to that it can be escalate.
Last modified 1yr ago