Spring Cloud Stream throws exception when converting a message to a manifold created object from JSON

[dakotahNorth]

Spring Cloud Stream can't automatically convert a manifold created object from JSON to POJO.

First the code with a POJO called Person ...

package com.example.helloworldspringcloudstream;

import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.function.Consumer;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class HelloWorldSpringCloudStreamApplication {

    public static void main(String... args) {
        SpringApplication.run(HelloWorldSpringCloudStreamApplication.class, args);

    }

    @Bean
    public Consumer<Person> exampleEventConsumer() {

        return event -> {
            System.out.println("Received event: " + event.getName());

        };

    }


}

And the Person POJO in Person.Java :

package com.example.helloworldspringcloudstream;

public class Person {

    private String name;

    public Person() { }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }


}

The above works as expected ... a JSON message that is sent into the broker is automically converted to a Person object.

Then, swapping in PersonManifold from JSON

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "$id": "http://com.example.helloworldspringcloudstream.PersonManifold.json",
  "title": "PersonManifold",
  "type": "object",
  "properties": {
    "name": { "type": "string" }
  }

}

And corresponding change in callback ...

    @Bean
    public Consumer<PersonManifold> exampleEventConsumer() {

        return event -> {
            System.out.println("Received event: " + event.getName());

        };

    }

And sending the same message as above, results in a failure to convert the message:

2024-04-16T11:25:30.927-04:00  WARN 47125 --- [HelloWorldSpringCloudStream] [pool-3-thread-1] c.s.s.c.s.b.i.JCSMPInboundChannelAdapter : Failed to consume a message from destination #P2P/QTMP/v:vzk7e2ikt35primary/scst/an/bbe6e438-7852-45c5-97f0-f8d1aa8bc466/plain/try-me - attempt 3
2024-04-16T11:25:30.933-04:00 ERROR 47125 --- [HelloWorldSpringCloudStream] [pool-3-thread-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@626467d0], failedMessage=GenericMessage [payload=byte[23], headers={solace_expiration=0, solace_destination=try-me, solace_replicationGroupMessageId=rmid1:30704-81ae9045b12-00000000-0000018d, deliveryAttempt=3, solace_isReply=false, solace_timeToLive=0, solace_receiveTimestamp=0, acknowledgmentCallback=com.solace.spring.cloud.stream.binder.inbound.acknowledge.JCSMPAcknowledgementCallback@676475b6, solace_discardIndication=false, solace_dmqEligible=false, solace_priority=-1, solace_redelivered=false, id=bd72984f-5f2c-d352-2710-85a3809c0e97, contentType=application/json, timestamp=1713281127830}]


Caused by: java.lang.RuntimeException: Receiver type 'byte[]' does not implement a method structurally compatible with method: public abstract java.lang.String com.example.helloworldspringcloudstream.PersonManifold.getName()
	at manifold.util.ReflectUtil.structuralCallByProxy(ReflectUtil.java:1605)
	at manifold.ext.rt.RuntimeMethods.lambda$null$7(RuntimeMethods.java:435)
	at jdk.proxy2/com.sun.proxy.jdk.proxy2.$ManProxy0.getName(Unknown Source)

[dakotahNorth]

Added -parameters to the compiler options and rebuilt ... still not working with Manifold JSON.

plugins {
    id 'java'
    id 'org.springframework.boot' version '3.2.4'
    id 'io.spring.dependency-management' version '1.1.4'
}

group = 'com.example.helloworldspringcloudstream'
version = '0.0.1-SNAPSHOT'

java {
    sourceCompatibility = '21'
}

repositories {
    mavenCentral()
    maven { url 'https://oss.sonatype.org/content/repositories/snapshots/' }
}


ext {
    springCloudVersion = '2023.0.1'
    manifoldVersion = '2024.1.12'
}


dependencies {
    implementation 'org.springframework.cloud:spring-cloud-stream'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.cloud:spring-cloud-stream-test-binder'

    implementation('com.solace.spring.cloud:spring-cloud-starter-stream-solace:5.0.0')

    implementation "systems.manifold:manifold-json-rt:${manifoldVersion}"
    annotationProcessor "systems.manifold:manifold-json:${manifoldVersion}"
    testAnnotationProcessor "systems.manifold:manifold-json:${manifoldVersion}"

}

compileJava {
    options.annotationProcessorPath = configurations.annotationProcessor
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
    }
}


tasks.withType(JavaCompile).configureEach {
    options.incremental = true
    options.compilerArgs += ['-Xplugin:Manifold']
    options.compilerArgs += ['-parameters']

}

tasks.named('test') {
    useJUnitPlatform()
}

[rsmckinney]

The problem is probably related to PersonManifold being an interface, Spring Cloud Stream can't create an instance of it.

[dakotahNorth]

Tried a little experiment ... I commented out the no-argument constructor for the Person object and an exception is thrown from the same method.

Within FunctionConfiguration handleMessageInternal both no-argument Person and PersonManifold are throwing an exception at the apply method.

      public void handleMessageInternal(Message<?> message) throws MessagingException {
	      Object result = functionInvocationWrapper.apply((Message<byte[]>) message);
	      if (result == null) {
		      logger.debug("Function execution resulted in null. No message will be sent");
		      return;
	      }
	      if (result instanceof Iterable<?> iterableResult) {
		      for (Object resultElement : iterableResult) {
			      this.doSendMessage(resultElement, message);
		      }
	      }
	      else if (ObjectUtils.isArray(result) && !(result instanceof byte[])) {
		      for (int i = 0; i < ((Object[]) result).length; i++) {
			      this.doSendMessage(((Object[]) result)[i], message);
		      }
	      }
	      else {
		      this.doSendMessage(result, message);
	      }
      }

This doesn't prove or disprove your comment ... but it does say that at the very least, Spring Cloud Stream is expecting a no-argument constructor.

[rsmckinney]

I'm pretty confident the issue with PersonManifold is due to the fact that it's an interface, spring needs a concrete class to construct.

I don't use spring, however I took a brief look and it appears you may be able to provide a custom binder so you can override handleMessage() and do whatever you like. That being the case, you could implement bindings for manifold-json, something like this:

Class<?> beanClass = get json interface from spring;
String jsonString = new String(jsonByteArray);
Object jsonObj = manifold.json.rt.Json.fromJson(jsonString);
Object obj = manifold.ext.rt.RuntimeMethods.coerce(jsonObj, beanClass); 
return obj;

With this in place your original exampleEventConsumer example should work.

[dakotahNorth]

Brilliant! Created the following classes

package com.example.helloworldspringcloudstream;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.util.MimeType;
import manifold.json.rt.Json;
import manifold.ext.rt.RuntimeMethods;

public class ManifoldJsonMessageConverter extends AbstractMessageConverter {

    public ManifoldJsonMessageConverter() {
        super(new MimeType("application", "json"));
    }

    @Override
    protected boolean supports(Class<?> clazz) {
        // This converter supports all types, but you might want to limit this
        return true;
    }

    @Override
    protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
        Object payload = message.getPayload();
        if (payload instanceof byte[]) {
            String jsonString = new String((byte[]) payload);
            Object jsonObj = Json.fromJson(jsonString);
            return RuntimeMethods.coerce(jsonObj, targetClass);
        }
        return null;
    }
}

and

package com.example.helloworldspringcloudstream;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MessageConverter;

import java.util.ArrayList;
import java.util.List;

@Configuration
public class CustomConverterConfig {

    @Bean
    public MessageConverter customMessageConverter() {
        List<MessageConverter> converters = new ArrayList<>();
        converters.add(new ManifoldJsonMessageConverter());
        return new CompositeMessageConverter(converters);
    }
}

By registering the custom converter as a bean, the above code automatically used it for message conversion.

[dakotahNorth]

Workaround provided above to use manifold with JSON.