Added example with Spring Integration ZeroMQ lib for Spring framework

This commit is contained in:
DaneliyaRI committed 2021-07-02 12:33:13 +07:00
1 parent 04b28b61c0
commit 27de365191
10 files changed
+403

No files matched your search

@@ -0,0 +1,15 @@
package com.eddn.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@SpringBootApplication
@EnableConfigurationProperties
public class EdcmApiJavaApplication {
public static void main(String[] args) {
SpringApplication.run(EdcmApiJavaApplication.class, args);
}
}
@@ -0,0 +1,54 @@
package com.eddn.example.core.zeromq;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.zeromq.channel.ZeroMqChannel;
import org.springframework.messaging.MessageHandler;
import org.zeromq.ZContext;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.zip.DataFormatException;
import java.util.zip.Inflater;
@Configuration
@EnableIntegration
public class ZeromqConfig {
@Bean
public ZContext zContext() {
return new ZContext();
}
@Bean(name = "zeroMqChannel")
ZeroMqChannel zeroMqPubSubChannel(ZContext context) {
ZeroMqChannel channel = new ZeroMqChannel(context, true);
channel.setConnectUrl("tcp://eddn.edcd.io:9500:9500");
return channel;
}
@Bean
@ServiceActivator(inputChannel = "zeroMqChannel")
public MessageHandler subscribe() {
return message -> {
byte[] output = new byte[256 * 1024];
byte[] payload = (byte[]) message.getPayload();
Inflater inflater = new Inflater();
inflater.setInput(payload);
ByteArrayOutputStream outputStream = new ByteArrayOutputStream(payload.length);
try {
int outputLength = inflater.inflate(output);
String outputString = new String(output, 0, outputLength, StandardCharsets.UTF_8);
System.out.println(outputString);
} catch (DataFormatException e) {
e.printStackTrace();
}
};
}
}
Whitespace-only changes.