diff --git a/mqtt-broker/pom.xml b/mqtt-broker/pom.xml
index 2293b02ff..8fe393e34 100644
--- a/mqtt-broker/pom.xml
+++ b/mqtt-broker/pom.xml
@@ -20,7 +20,7 @@
pulsar-protocol-handler-mqtt-parent
io.streamnative.pulsar.handlers
- 3.4.0-SNAPSHOT
+ 4.0.14.1
4.0.0
pulsar-protocol-handler-mqtt
diff --git a/mqtt-common/pom.xml b/mqtt-common/pom.xml
index 707ea52b6..d0a184425 100644
--- a/mqtt-common/pom.xml
+++ b/mqtt-common/pom.xml
@@ -20,7 +20,7 @@
pulsar-protocol-handler-mqtt-parent
io.streamnative.pulsar.handlers
- 3.4.0-SNAPSHOT
+ 4.0.14.1
4.0.0
pulsar-protocol-handler-mqtt-common
diff --git a/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java
index 461f6f249..c01c19260 100644
--- a/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java
+++ b/mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/utils/ConfigurationUtils.java
@@ -16,8 +16,6 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.lang.String.format;
-import static org.apache.pulsar.common.util.FieldParser.setEmptyValue;
-import static org.apache.pulsar.common.util.FieldParser.value;
import com.google.common.base.Joiner;
import java.io.FileInputStream;
import java.io.IOException;
@@ -26,11 +24,14 @@
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.common.configuration.FieldContext;
@@ -155,6 +156,159 @@ public static void update(Map properties, T obj) throws Ille
});
}
+ private static Object value(String value, Field field) {
+ checkNotNull(field);
+ Type fieldType = field.getGenericType();
+ if (fieldType instanceof ParameterizedType) {
+ Type[] actualTypes = ((ParameterizedType) fieldType).getActualTypeArguments();
+ if (field.getType().equals(List.class)) {
+ return stringToList(value, getRawClass(actualTypes[0], field));
+ } else if (field.getType().equals(Set.class)) {
+ return stringToSet(value, getRawClass(actualTypes[0], field));
+ } else if (field.getType().equals(Map.class)) {
+ return stringToMap(value, getRawClass(actualTypes[0], field), getRawClass(actualTypes[1], field));
+ } else if (field.getType().equals(Optional.class)) {
+ Type typeClazz = actualTypes[0];
+ if (typeClazz instanceof ParameterizedType) {
+ throw new IllegalArgumentException(format("unsupported non-primitive Optional<%s> for %s",
+ typeClazz.getClass(), field.getName()));
+ }
+ return Optional.ofNullable(convert(value, getRawClass(typeClazz, field)));
+ }
+ throw new IllegalArgumentException(format("unsupported field-type %s for %s",
+ field.getType(), field.getName()));
+ }
+ return convert(value, field.getType());
+ }
+
+ private static void setEmptyValue(String value, Field field, T obj)
+ throws IllegalArgumentException, IllegalAccessException {
+ checkNotNull(field);
+ Type fieldType = field.getGenericType();
+ if (fieldType instanceof ParameterizedType) {
+ if (field.getType().equals(List.class)) {
+ field.set(obj, new ArrayList<>());
+ } else if (field.getType().equals(Set.class)) {
+ field.set(obj, new LinkedHashSet<>());
+ } else if (field.getType().equals(Optional.class)) {
+ field.set(obj, Optional.empty());
+ } else {
+ throw new IllegalArgumentException(format("unsupported field-type %s for %s",
+ field.getType(), field.getName()));
+ }
+ } else if (Number.class.isAssignableFrom(field.getType())) {
+ field.set(obj, null);
+ }
+ }
+
+ private static Object convert(Object value, Class> targetType) {
+ checkNotNull(targetType);
+ if (value == null) {
+ return null;
+ }
+ Class> wrappedType = wrap(targetType);
+ if (wrappedType.isAssignableFrom(value.getClass())) {
+ return wrappedType.cast(value);
+ }
+ if (wrappedType.isEnum()) {
+ return stringToEnum(value.toString(), wrappedType);
+ }
+ if (wrappedType.equals(Integer.class)) {
+ return stringToInteger(value.toString());
+ } else if (wrappedType.equals(Long.class)) {
+ return Long.valueOf(value.toString().trim());
+ } else if (wrappedType.equals(Double.class)) {
+ return stringToDouble(value.toString());
+ } else if (wrappedType.equals(Float.class)) {
+ return Float.valueOf(value.toString().trim());
+ } else if (wrappedType.equals(Boolean.class)) {
+ return Boolean.valueOf(value.toString());
+ }
+ throw new UnsupportedOperationException(format("Cannot convert from %s to %s. Requested converter does not"
+ + " exist.", value.getClass().getName(), wrappedType.getName()));
+ }
+
+ private static Class> getRawClass(Type type, Field field) {
+ if (type instanceof Class) {
+ return (Class>) type;
+ }
+ throw new IllegalArgumentException(format("unsupported field-type %s for %s", type, field.getName()));
+ }
+
+ private static Class> wrap(Class> clazz) {
+ if (clazz.equals(Integer.TYPE)) {
+ return Integer.class;
+ } else if (clazz.equals(Float.TYPE)) {
+ return Float.class;
+ } else if (clazz.equals(Double.TYPE)) {
+ return Double.class;
+ } else if (clazz.equals(Long.TYPE)) {
+ return Long.class;
+ } else if (clazz.equals(Boolean.TYPE)) {
+ return Boolean.class;
+ }
+ return clazz;
+ }
+
+ private static Integer stringToInteger(String value) {
+ String trimmedValue = value.trim();
+ if (StringUtils.isEmpty(trimmedValue)) {
+ return null;
+ }
+ return Integer.valueOf(trimmedValue);
+ }
+
+ private static Double stringToDouble(String value) {
+ String trimmedValue = value.trim();
+ if (StringUtils.isEmpty(trimmedValue)) {
+ return null;
+ }
+ return Double.valueOf(trimmedValue);
+ }
+
+ private static List stringToList(String value, Class type) {
+ if (value == null) {
+ return null;
+ }
+ return Arrays.stream(value.trim().split(LISTENER_DEL))
+ .map(item -> type.cast(convert(item.trim(), type)))
+ .collect(Collectors.toList());
+ }
+
+ private static Set stringToSet(String value, Class type) {
+ if (value == null) {
+ return null;
+ }
+ return Arrays.stream(value.trim().split(LISTENER_DEL))
+ .map(item -> type.cast(convert(item.trim(), type)))
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ private static Map stringToMap(String value, Class keyType, Class valueType) {
+ if (value == null) {
+ return null;
+ }
+ return Arrays.stream(value.trim().split(LISTENER_DEL))
+ .map(item -> item.trim().split("=", 2))
+ .peek(parts -> checkArgument(parts.length == 2,
+ value + " map-value is not in correct format key1=value,key2=value2"))
+ .collect(Collectors.toMap(
+ parts -> keyType.cast(convert(parts[0].trim(), keyType)),
+ parts -> valueType.cast(convert(parts[1].trim(), valueType)),
+ (oldValue, newValue) -> newValue));
+ }
+
+ private static Object stringToEnum(String value, Class> enumType) {
+ String trimmedValue = value.trim();
+ for (Object enumConstant : enumType.getEnumConstants()) {
+ Enum> enumValue = (Enum>) enumConstant;
+ if (enumValue.toString().equals(trimmedValue) || enumValue.name().equals(trimmedValue)) {
+ return enumValue;
+ }
+ }
+ throw new RuntimeException(format("Invalid value '%s' for enum %s", value, enumType));
+ }
+
public static int getListenerPort(String listener) {
checkArgument(listener.matches(LISTENER_PATTERN), "listener not match pattern");
diff --git a/mqtt-proxy/pom.xml b/mqtt-proxy/pom.xml
index 47fdb6df7..852aba200 100644
--- a/mqtt-proxy/pom.xml
+++ b/mqtt-proxy/pom.xml
@@ -20,7 +20,7 @@
pulsar-protocol-handler-mqtt-parent
io.streamnative.pulsar.handlers
- 3.4.0-SNAPSHOT
+ 4.0.14.1
4.0.0
pulsar-protocol-handler-mqtt-proxy
diff --git a/pom.xml b/pom.xml
index 5cfbdb037..01c6c5288 100644
--- a/pom.xml
+++ b/pom.xml
@@ -19,7 +19,7 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
io.streamnative.pulsar.handlers
pulsar-protocol-handler-mqtt-parent
- 3.4.0-SNAPSHOT
+ 4.0.14.1
StreamNative :: Pulsar Protocol Handler :: MoP Parent
Parent for MQTT on Pulsar implemented using Pulsar Protocol Handler.
@@ -51,8 +51,8 @@
2.22.0
6.14.3
4.0.2
- 4.0.10.9
- 4.0.10.9
+ 4.0.14.1
+ 4.0.14.1
2.18.0
1.16
1.2.2
diff --git a/tests/pom.xml b/tests/pom.xml
index ae2736668..8e6de32d4 100644
--- a/tests/pom.xml
+++ b/tests/pom.xml
@@ -20,7 +20,7 @@
pulsar-protocol-handler-mqtt-parent
io.streamnative.pulsar.handlers
- 3.4.0-SNAPSHOT
+ 4.0.14.1
4.0.0
pulsar-protocol-handler-mqtt-tests
diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java
index 47a689fa2..28ca3d293 100644
--- a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java
+++ b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/base/MQTTProtocolHandlerTestBase.java
@@ -142,6 +142,7 @@ protected MQTTCommonConfiguration initConfig() throws Exception{
mqtt.setDefaultNumberOfNamespaceBundles(4);
mqtt.setZookeeperServers("localhost:2181");
mqtt.setConfigurationStoreServers("localhost:3181");
+ mqtt.setTopicLevelPoliciesEnabled(false);
mqtt.setAuthenticationEnabled(false);
mqtt.setAuthorizationEnabled(false);
@@ -390,6 +391,8 @@ protected void startBroker(MQTTCommonConfiguration conf) throws Exception {
mqttBrokerPort, mqttBrokerTlsPort, mqttBrokerTlsPskPort,
mqttProxyPort, mqttProxyTlsPort, mqttProxyTlsPskPort);
ConfigurationUtils.extractFieldToProperties(conf);
+ conf.getProperties().remove("managedLedgerDataReadPriority");
+ conf.getProperties().remove("managedLedgerOffloadedReadPriority");
setTLSConf(conf);
mockZooKeeper.setSessionId(pulsarServiceList.size() + 1);
this.pulsarServiceList.add(doStartBroker(conf));