Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Selector header name is exposed for configuration

Browse Source 
Issue: SPR-16732
This commit is contained in:
Rossen Stoyanchev
2018-04-17 00:20:12 -04:00
parent 8748ba4c4b
commit 246a6db1ca
8 changed files with 144 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -43,6 +43,7 @@ import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.PathMatcher;
import org.springframework.util.StringUtils;
/**
* Implementation of {@link SubscriptionRegistry} that stores subscriptions
@@ -113,24 +114,25 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
}
/**
* Configure the name of a selector header that a subscription message can
* have in order to filter messages based on their headers. The value of the
* header can use Spring EL expressions against message headers.
* <p>For example the following expression expects a header called "foo" to
* have the value "bar":
* Configure the name of a header that a subscription message can have for
* the purpose of filtering messages matched to the subscription. The header
* value is expected to be a Spring EL boolean expression to be applied to
* the headers of messages matched to the subscription.
* <p>For example:
* <pre>
* headers.foo == 'bar'
* </pre>
* <p>By default this is set to "selector".
* <p>By default this is set to "selector". You can set it to a different
* name, or to {@code null} to turn off support for a selector header.
* @param selectorHeaderName the name to use for a selector header
* @since 4.2
*/
public void setSelectorHeaderName(String selectorHeaderName) {
Assert.notNull(selectorHeaderName, "'selectorHeaderName' must not be null");
this.selectorHeaderName = selectorHeaderName;
this.selectorHeaderName = StringUtils.hasText(selectorHeaderName) ? selectorHeaderName : null;
}
/**
* Return the name for the selector header.
* Return the name for the selector header name.
* @since 4.2
*/
public String getSelectorHeaderName() {
@@ -142,25 +144,31 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
protected void addSubscriptionInternal(
String sessionId, String subsId, String destination, Message<?> message) {
Expression expression = getSelectorExpression(message.getHeaders());
this.subscriptionRegistry.addSubscription(sessionId, subsId, destination, expression);
this.destinationCache.updateAfterNewSubscription(destination, sessionId, subsId);
}
private Expression getSelectorExpression(MessageHeaders headers) {
Expression expression = null;
MessageHeaders headers = message.getHeaders();
String selector = SimpMessageHeaderAccessor.getFirstNativeHeader(getSelectorHeaderName(), headers);
if (selector != null) {
try {
expression = this.expressionParser.parseExpression(selector);
this.selectorHeaderInUse = true;
if (logger.isTraceEnabled()) {
logger.trace("Subscription selector: [" + selector + "]");
if (getSelectorHeaderName() != null) {
String selector = SimpMessageHeaderAccessor.getFirstNativeHeader(getSelectorHeaderName(), headers);
if (selector != null) {
try {
expression = this.expressionParser.parseExpression(selector);
this.selectorHeaderInUse = true;
if (logger.isTraceEnabled()) {
logger.trace("Subscription selector: [" + selector + "]");
}
}
}
catch (Throwable ex) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to parse selector: " + selector, ex);
catch (Throwable ex) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to parse selector: " + selector, ex);
}
}
}
}
this.subscriptionRegistry.addSubscription(sessionId, subsId, destination, expression);
this.destinationCache.updateAfterNewSubscription(destination, sessionId, subsId);
return expression;
}
@Override

44
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -51,23 +51,27 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private final Map<String, SessionInfo> sessions = new ConcurrentHashMap<String, SessionInfo>();
private SubscriptionRegistry subscriptionRegistry;
private PathMatcher pathMatcher;
private Integer cacheLimit;
private String selectorHeaderName = "selector";
private TaskScheduler taskScheduler;
private long[] heartbeatValue;
private ScheduledFuture<?> heartbeatFuture;
private MessageHeaderInitializer headerInitializer;
private SubscriptionRegistry subscriptionRegistry;
private final Map<String, SessionInfo> sessions = new ConcurrentHashMap<String, SessionInfo>();
private ScheduledFuture<?> heartbeatFuture;
/**
* Create a SimpleBrokerMessageHandler instance with the given message channels
* and destination prefixes.
@@ -96,12 +100,40 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
this.subscriptionRegistry = subscriptionRegistry;
initPathMatcherToUse();
initCacheLimitToUse();
initSelectorHeaderNameToUse();
}
public SubscriptionRegistry getSubscriptionRegistry() {
return this.subscriptionRegistry;
}
/**
* Configure the name of a header that a subscription message can have for
* the purpose of filtering messages matched to the subscription. The header
* value is expected to be a Spring EL boolean expression to be applied to
* the headers of messages matched to the subscription.
* <p>For example:
* <pre>
* headers.foo == 'bar'
* </pre>
* <p>By default this is set to "selector". You can set it to a different
* name, or to {@code null} to turn off support for a selector header.
* @param selectorHeaderName the name to use for a selector header
* @since 4.3.17
* @see #setSubscriptionRegistry
* @see DefaultSubscriptionRegistry#setSelectorHeaderName(String)
*/
public void setSelectorHeaderName(String selectorHeaderName) {
this.selectorHeaderName = selectorHeaderName;
initSelectorHeaderNameToUse();
}
private void initSelectorHeaderNameToUse() {
if (this.subscriptionRegistry instanceof DefaultSubscriptionRegistry) {
((DefaultSubscriptionRegistry) this.subscriptionRegistry).setSelectorHeaderName(this.selectorHeaderName);
}
}
/**
* When configured, the given PathMatcher is passed down to the underlying
* SubscriptionRegistry to use for matching destination to subscriptions.

23
spring-messaging/src/main/java/org/springframework/messaging/simp/config/SimpleBrokerRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -33,6 +33,8 @@ public class SimpleBrokerRegistration extends AbstractBrokerRegistration {
private long[] heartbeat;
private String selectorHeaderName = "selector";
public SimpleBrokerRegistration(SubscribableChannel inChannel, MessageChannel outChannel, String[] prefixes) {
super(inChannel, outChannel, prefixes);
@@ -65,6 +67,24 @@ public class SimpleBrokerRegistration extends AbstractBrokerRegistration {
return this;
}
/**
* Configure the name of a header that a subscription message can have for
* the purpose of filtering messages matched to the subscription. The header
* value is expected to be a Spring EL boolean expression to be applied to
* the headers of messages matched to the subscription.
* <p>For example:
* <pre>
* headers.foo == 'bar'
* </pre>
* <p>By default this is set to "selector". You can set it to a different
* name, or to {@code null} to turn off support for a selector header.
* @param selectorHeaderName the name to use for a selector header
* @since 4.3.17
*/
public void setSelectorHeaderName(String selectorHeaderName) {
this.selectorHeaderName = selectorHeaderName;
}
@Override
protected SimpleBrokerMessageHandler getMessageHandler(SubscribableChannel brokerChannel) {
@@ -76,6 +96,7 @@ public class SimpleBrokerRegistration extends AbstractBrokerRegistration {
if (this.heartbeat != null) {
handler.setHeartbeatValue(this.heartbeat);
}
handler.setSelectorHeaderName(this.selectorHeaderName);
return handler;
}