ActiveMQ Failover transport over Spring Stomp Broker Relay
.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty{ height:90px;width:728px;box-sizing:border-box;
}
Did anyone succeeded using failover transport utilizing active/standby Active MQ (Amazon MQ to be precise) when configuring websocket stomp broker relay ?
I did create following TcpClient, but it implements round-robin algorithm over given endpoints, which causes some errors when actually standby instance is being selected:
Transport failure: epoll_ctl(..) failed: No such file or directory
Here's my current implementation:
public class StompTcpFactory implements NetStreams.TcpClientFactory<Message<byte>, Message<byte>> {
private final Environment environment = new Environment(new SynchronousDispatcherConfigReader());
private final List<URI> endpoints;
private final boolean isSsl;
public StompTcpFactory(List<String> endpoints) {
this.endpoints = endpoints.stream()
.map(e -> contains(e, "://") ? e : "fake://" + e)
.map(URI::create)
.collect(toList());
isSsl = this.endpoints.stream().anyMatch(StompTcpFactory::isSsl);
boolean anyNotSsl = this.endpoints.stream().anyMatch(not(StompTcpFactory::isSsl));
if (isSsl && anyNotSsl)
throw new IllegalArgumentException("Cannot configure STOMP to use SSL and regular connections at the same time: " + endpoints);
}
@Override
public Spec.TcpClientSpec<Message<byte>, Message<byte>> apply(Spec.TcpClientSpec<Message<byte>, Message<byte>> tcpClientSpec) {
return tcpClientSpec
.env(environment)
.codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder()))
.ssl(isSsl ? new SslOptions() : null)
.connect(new InetSocketAddressSupplier(endpoints));
}
private static boolean isSsl(URI endpoint) {
return containsIgnoreCase(endpoint.getScheme(), "ssl");
}
private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
@Override
public ReactorConfiguration read() {
return new ReactorConfiguration(emptyList(), "sync", new Properties());
}
}
}
with following supplier:
public class InetSocketAddressSupplier implements Supplier<InetSocketAddress> {
private static final AtomicInteger counter = new AtomicInteger(0);
private final Logger logger = LoggerFactory.getLogger(getClass());
private final List<URI> endpoints;
public InetSocketAddressSupplier(List<URI> endpoints) {
this.endpoints = endpoints;
}
@Override
public InetSocketAddress get() {
int endpointIndex = counter.getAndUpdate(i -> ++i % endpoints.size());
URI endpoint = endpoints.get(endpointIndex);
logger.info("nnConnecting to broker[{}]: {}:{}nn", endpointIndex, endpoint.getHost(), endpoint.getPort());
return new InetSocketAddress(endpoint.getHost(), endpoint.getPort());
}
}
And here's the broker configuration:
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
String environment = configService.getString(ENVIRONMENT);
String user = configService.getString(WEBSOCKET_EXTERNAL_BROKER_USER);
String password = configService.getString(WEBSOCKET_EXTERNAL_BROKER_PASSWORD);
registry
.enableStompBrokerRelay("/queue", "/topic")
.setRelayHost(UNUSED_RELAY_HOST)
.setClientLogin(user)
.setClientPasscode(password)
.setSystemLogin(user)
.setSystemPasscode(password)
.setUserDestinationBroadcast(format("/topic/%s-unresolved-user-destination", environment))
.setUserRegistryBroadcast(format("/topic/%s-simp-user-registry", environment))
.setTcpClient(createTcpClient());
registry.setApplicationDestinationPrefixes(format("/%s-websocket-app", environment));
}
private TcpOperations<byte> createTcpClient() {
List<String> endpoints = asList(configService.getStringArray(WEBSOCKET_EXTERNAL_BROKER_ENDPOINTS));
logger.info("Configuring websocket brokers to: {}", endpoints);
return new Reactor2TcpClient<>(new StompTcpFactory(endpoints));
}
websocket activemq spring-websocket stomp java-websocket
add a comment |
Did anyone succeeded using failover transport utilizing active/standby Active MQ (Amazon MQ to be precise) when configuring websocket stomp broker relay ?
I did create following TcpClient, but it implements round-robin algorithm over given endpoints, which causes some errors when actually standby instance is being selected:
Transport failure: epoll_ctl(..) failed: No such file or directory
Here's my current implementation:
public class StompTcpFactory implements NetStreams.TcpClientFactory<Message<byte>, Message<byte>> {
private final Environment environment = new Environment(new SynchronousDispatcherConfigReader());
private final List<URI> endpoints;
private final boolean isSsl;
public StompTcpFactory(List<String> endpoints) {
this.endpoints = endpoints.stream()
.map(e -> contains(e, "://") ? e : "fake://" + e)
.map(URI::create)
.collect(toList());
isSsl = this.endpoints.stream().anyMatch(StompTcpFactory::isSsl);
boolean anyNotSsl = this.endpoints.stream().anyMatch(not(StompTcpFactory::isSsl));
if (isSsl && anyNotSsl)
throw new IllegalArgumentException("Cannot configure STOMP to use SSL and regular connections at the same time: " + endpoints);
}
@Override
public Spec.TcpClientSpec<Message<byte>, Message<byte>> apply(Spec.TcpClientSpec<Message<byte>, Message<byte>> tcpClientSpec) {
return tcpClientSpec
.env(environment)
.codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder()))
.ssl(isSsl ? new SslOptions() : null)
.connect(new InetSocketAddressSupplier(endpoints));
}
private static boolean isSsl(URI endpoint) {
return containsIgnoreCase(endpoint.getScheme(), "ssl");
}
private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
@Override
public ReactorConfiguration read() {
return new ReactorConfiguration(emptyList(), "sync", new Properties());
}
}
}
with following supplier:
public class InetSocketAddressSupplier implements Supplier<InetSocketAddress> {
private static final AtomicInteger counter = new AtomicInteger(0);
private final Logger logger = LoggerFactory.getLogger(getClass());
private final List<URI> endpoints;
public InetSocketAddressSupplier(List<URI> endpoints) {
this.endpoints = endpoints;
}
@Override
public InetSocketAddress get() {
int endpointIndex = counter.getAndUpdate(i -> ++i % endpoints.size());
URI endpoint = endpoints.get(endpointIndex);
logger.info("nnConnecting to broker[{}]: {}:{}nn", endpointIndex, endpoint.getHost(), endpoint.getPort());
return new InetSocketAddress(endpoint.getHost(), endpoint.getPort());
}
}
And here's the broker configuration:
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
String environment = configService.getString(ENVIRONMENT);
String user = configService.getString(WEBSOCKET_EXTERNAL_BROKER_USER);
String password = configService.getString(WEBSOCKET_EXTERNAL_BROKER_PASSWORD);
registry
.enableStompBrokerRelay("/queue", "/topic")
.setRelayHost(UNUSED_RELAY_HOST)
.setClientLogin(user)
.setClientPasscode(password)
.setSystemLogin(user)
.setSystemPasscode(password)
.setUserDestinationBroadcast(format("/topic/%s-unresolved-user-destination", environment))
.setUserRegistryBroadcast(format("/topic/%s-simp-user-registry", environment))
.setTcpClient(createTcpClient());
registry.setApplicationDestinationPrefixes(format("/%s-websocket-app", environment));
}
private TcpOperations<byte> createTcpClient() {
List<String> endpoints = asList(configService.getStringArray(WEBSOCKET_EXTERNAL_BROKER_ENDPOINTS));
logger.info("Configuring websocket brokers to: {}", endpoints);
return new Reactor2TcpClient<>(new StompTcpFactory(endpoints));
}
websocket activemq spring-websocket stomp java-websocket
add a comment |
Did anyone succeeded using failover transport utilizing active/standby Active MQ (Amazon MQ to be precise) when configuring websocket stomp broker relay ?
I did create following TcpClient, but it implements round-robin algorithm over given endpoints, which causes some errors when actually standby instance is being selected:
Transport failure: epoll_ctl(..) failed: No such file or directory
Here's my current implementation:
public class StompTcpFactory implements NetStreams.TcpClientFactory<Message<byte>, Message<byte>> {
private final Environment environment = new Environment(new SynchronousDispatcherConfigReader());
private final List<URI> endpoints;
private final boolean isSsl;
public StompTcpFactory(List<String> endpoints) {
this.endpoints = endpoints.stream()
.map(e -> contains(e, "://") ? e : "fake://" + e)
.map(URI::create)
.collect(toList());
isSsl = this.endpoints.stream().anyMatch(StompTcpFactory::isSsl);
boolean anyNotSsl = this.endpoints.stream().anyMatch(not(StompTcpFactory::isSsl));
if (isSsl && anyNotSsl)
throw new IllegalArgumentException("Cannot configure STOMP to use SSL and regular connections at the same time: " + endpoints);
}
@Override
public Spec.TcpClientSpec<Message<byte>, Message<byte>> apply(Spec.TcpClientSpec<Message<byte>, Message<byte>> tcpClientSpec) {
return tcpClientSpec
.env(environment)
.codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder()))
.ssl(isSsl ? new SslOptions() : null)
.connect(new InetSocketAddressSupplier(endpoints));
}
private static boolean isSsl(URI endpoint) {
return containsIgnoreCase(endpoint.getScheme(), "ssl");
}
private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
@Override
public ReactorConfiguration read() {
return new ReactorConfiguration(emptyList(), "sync", new Properties());
}
}
}
with following supplier:
public class InetSocketAddressSupplier implements Supplier<InetSocketAddress> {
private static final AtomicInteger counter = new AtomicInteger(0);
private final Logger logger = LoggerFactory.getLogger(getClass());
private final List<URI> endpoints;
public InetSocketAddressSupplier(List<URI> endpoints) {
this.endpoints = endpoints;
}
@Override
public InetSocketAddress get() {
int endpointIndex = counter.getAndUpdate(i -> ++i % endpoints.size());
URI endpoint = endpoints.get(endpointIndex);
logger.info("nnConnecting to broker[{}]: {}:{}nn", endpointIndex, endpoint.getHost(), endpoint.getPort());
return new InetSocketAddress(endpoint.getHost(), endpoint.getPort());
}
}
And here's the broker configuration:
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
String environment = configService.getString(ENVIRONMENT);
String user = configService.getString(WEBSOCKET_EXTERNAL_BROKER_USER);
String password = configService.getString(WEBSOCKET_EXTERNAL_BROKER_PASSWORD);
registry
.enableStompBrokerRelay("/queue", "/topic")
.setRelayHost(UNUSED_RELAY_HOST)
.setClientLogin(user)
.setClientPasscode(password)
.setSystemLogin(user)
.setSystemPasscode(password)
.setUserDestinationBroadcast(format("/topic/%s-unresolved-user-destination", environment))
.setUserRegistryBroadcast(format("/topic/%s-simp-user-registry", environment))
.setTcpClient(createTcpClient());
registry.setApplicationDestinationPrefixes(format("/%s-websocket-app", environment));
}
private TcpOperations<byte> createTcpClient() {
List<String> endpoints = asList(configService.getStringArray(WEBSOCKET_EXTERNAL_BROKER_ENDPOINTS));
logger.info("Configuring websocket brokers to: {}", endpoints);
return new Reactor2TcpClient<>(new StompTcpFactory(endpoints));
}
websocket activemq spring-websocket stomp java-websocket
Did anyone succeeded using failover transport utilizing active/standby Active MQ (Amazon MQ to be precise) when configuring websocket stomp broker relay ?
I did create following TcpClient, but it implements round-robin algorithm over given endpoints, which causes some errors when actually standby instance is being selected:
Transport failure: epoll_ctl(..) failed: No such file or directory
Here's my current implementation:
public class StompTcpFactory implements NetStreams.TcpClientFactory<Message<byte>, Message<byte>> {
private final Environment environment = new Environment(new SynchronousDispatcherConfigReader());
private final List<URI> endpoints;
private final boolean isSsl;
public StompTcpFactory(List<String> endpoints) {
this.endpoints = endpoints.stream()
.map(e -> contains(e, "://") ? e : "fake://" + e)
.map(URI::create)
.collect(toList());
isSsl = this.endpoints.stream().anyMatch(StompTcpFactory::isSsl);
boolean anyNotSsl = this.endpoints.stream().anyMatch(not(StompTcpFactory::isSsl));
if (isSsl && anyNotSsl)
throw new IllegalArgumentException("Cannot configure STOMP to use SSL and regular connections at the same time: " + endpoints);
}
@Override
public Spec.TcpClientSpec<Message<byte>, Message<byte>> apply(Spec.TcpClientSpec<Message<byte>, Message<byte>> tcpClientSpec) {
return tcpClientSpec
.env(environment)
.codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder()))
.ssl(isSsl ? new SslOptions() : null)
.connect(new InetSocketAddressSupplier(endpoints));
}
private static boolean isSsl(URI endpoint) {
return containsIgnoreCase(endpoint.getScheme(), "ssl");
}
private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
@Override
public ReactorConfiguration read() {
return new ReactorConfiguration(emptyList(), "sync", new Properties());
}
}
}
with following supplier:
public class InetSocketAddressSupplier implements Supplier<InetSocketAddress> {
private static final AtomicInteger counter = new AtomicInteger(0);
private final Logger logger = LoggerFactory.getLogger(getClass());
private final List<URI> endpoints;
public InetSocketAddressSupplier(List<URI> endpoints) {
this.endpoints = endpoints;
}
@Override
public InetSocketAddress get() {
int endpointIndex = counter.getAndUpdate(i -> ++i % endpoints.size());
URI endpoint = endpoints.get(endpointIndex);
logger.info("nnConnecting to broker[{}]: {}:{}nn", endpointIndex, endpoint.getHost(), endpoint.getPort());
return new InetSocketAddress(endpoint.getHost(), endpoint.getPort());
}
}
And here's the broker configuration:
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
String environment = configService.getString(ENVIRONMENT);
String user = configService.getString(WEBSOCKET_EXTERNAL_BROKER_USER);
String password = configService.getString(WEBSOCKET_EXTERNAL_BROKER_PASSWORD);
registry
.enableStompBrokerRelay("/queue", "/topic")
.setRelayHost(UNUSED_RELAY_HOST)
.setClientLogin(user)
.setClientPasscode(password)
.setSystemLogin(user)
.setSystemPasscode(password)
.setUserDestinationBroadcast(format("/topic/%s-unresolved-user-destination", environment))
.setUserRegistryBroadcast(format("/topic/%s-simp-user-registry", environment))
.setTcpClient(createTcpClient());
registry.setApplicationDestinationPrefixes(format("/%s-websocket-app", environment));
}
private TcpOperations<byte> createTcpClient() {
List<String> endpoints = asList(configService.getStringArray(WEBSOCKET_EXTERNAL_BROKER_ENDPOINTS));
logger.info("Configuring websocket brokers to: {}", endpoints);
return new Reactor2TcpClient<>(new StompTcpFactory(endpoints));
}
websocket activemq spring-websocket stomp java-websocket
websocket activemq spring-websocket stomp java-websocket
asked Nov 22 '18 at 8:33
emberember
33
33
add a comment |
add a comment |
0
active
oldest
votes
Your Answer
StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");
StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);
StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});
function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});
}
});
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53426755%2factivemq-failover-transport-over-spring-stomp-broker-relay%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
0
active
oldest
votes
0
active
oldest
votes
active
oldest
votes
active
oldest
votes
Thanks for contributing an answer to Stack Overflow!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53426755%2factivemq-failover-transport-over-spring-stomp-broker-relay%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown