删除已经废弃的功能
This commit is contained in:
zhou-hao
parent
32d96332bc
commit
57d2f5fdc9
|
|
@ -1,21 +0,0 @@
|
|||
package org.jetlinks.community.gateway;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.Setter;
|
||||
import org.jetlinks.core.message.codec.EncodedMessage;
|
||||
import org.jetlinks.core.message.codec.Transport;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
@AllArgsConstructor
|
||||
class DefaultTopicMessage implements TopicMessage {
|
||||
private String topic;
|
||||
|
||||
private EncodedMessage message;
|
||||
|
||||
public DefaultTopicMessage(String topic,Object message){
|
||||
this.topic=topic;
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -15,17 +15,6 @@ import java.util.Optional;
|
|||
|
||||
public class DeviceMessageUtils {
|
||||
|
||||
@SuppressWarnings("all")
|
||||
public static Optional<DeviceMessage> convert(TopicMessage message){
|
||||
Object nativeMessage = message.convertMessage();
|
||||
if (nativeMessage instanceof DeviceMessage) {
|
||||
return Optional.of((DeviceMessage)nativeMessage);
|
||||
} else if (nativeMessage instanceof Map) {
|
||||
return MessageType.convertMessage(((Map<String, Object>) nativeMessage));
|
||||
}
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
public static Optional<DeviceMessage> convert(TopicPayload message) {
|
||||
return Optional.of(message.decode(DeviceMessage.class));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,66 +0,0 @@
|
|||
package org.jetlinks.community.gateway;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import io.netty.buffer.ByteBufUtil;
|
||||
import org.jetlinks.core.message.codec.EncodedMessage;
|
||||
import org.jetlinks.supports.utils.MqttTopicUtils;
|
||||
|
||||
import javax.annotation.Nonnull;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Map;
|
||||
|
||||
@Deprecated
|
||||
public interface TopicMessage {
|
||||
|
||||
/**
|
||||
* 主题: 格式为: /group/1/user/1, 支持通配符: **(多层路径),*(单层路径)
|
||||
*
|
||||
* <pre>
|
||||
* /group/** , /group/下的全部topic.包括子目录
|
||||
* /group/1/* , /group/1/下的topic. 不包括子目录
|
||||
* </pre>
|
||||
*
|
||||
* @return topic
|
||||
*/
|
||||
@Nonnull
|
||||
String getTopic();
|
||||
|
||||
/**
|
||||
* @return 已编码的消息
|
||||
* @see org.jetlinks.core.message.codec.MqttMessage
|
||||
*/
|
||||
@Nonnull
|
||||
EncodedMessage getMessage();
|
||||
|
||||
default Map<String, String> getTopicVars(String pattern) {
|
||||
return MqttTopicUtils.getPathVariables(pattern, getTopic());
|
||||
}
|
||||
|
||||
default Object convertMessage() {
|
||||
if (getMessage() instanceof EncodableMessage) {
|
||||
return ((EncodableMessage) getMessage()).getNativePayload();
|
||||
}
|
||||
byte[] payload = getMessage().payloadAsBytes();
|
||||
//maybe json
|
||||
if (/* { }*/(payload[0] == 123 && payload[payload.length - 1] == 125)
|
||||
|| /* [ ] */(payload[0] == 91 && payload[payload.length - 1] == 93)
|
||||
) {
|
||||
return JSON.parseObject(new String(payload));
|
||||
}
|
||||
if (ByteBufUtil.isText(getMessage().getPayload(), StandardCharsets.UTF_8)) {
|
||||
return getMessage().payloadAsString();
|
||||
}
|
||||
return payload;
|
||||
}
|
||||
|
||||
static TopicMessage of(String topic, EncodedMessage message) {
|
||||
return new DefaultTopicMessage(topic, message);
|
||||
}
|
||||
|
||||
static TopicMessage of(String topic, Object payload) {
|
||||
if (payload instanceof EncodedMessage) {
|
||||
return of(topic, ((EncodedMessage) payload));
|
||||
}
|
||||
return of(topic, EncodableMessage.of(payload));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
package org.jetlinks.community.gateway;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import org.jetlinks.core.NativePayload;
|
||||
import org.jetlinks.core.Payload;
|
||||
import org.jetlinks.core.event.TopicPayload;
|
||||
import org.jetlinks.core.message.codec.EncodedMessage;
|
||||
|
||||
import javax.annotation.Nonnull;
|
||||
|
||||
public class TopicMessageWrap implements TopicMessage {
|
||||
|
||||
private String topic;
|
||||
|
||||
private EncodedMessage message;
|
||||
|
||||
public static TopicMessageWrap wrap(TopicPayload topicPayload) {
|
||||
Payload payload = topicPayload.getPayload();
|
||||
TopicMessageWrap wrap = new TopicMessageWrap();
|
||||
wrap.topic = topicPayload.getTopic();
|
||||
if (payload instanceof NativePayload) {
|
||||
wrap.message = new EncodableMessage() {
|
||||
@Override
|
||||
public Object getNativePayload() {
|
||||
return ((NativePayload<?>) payload).getNativeObject();
|
||||
}
|
||||
|
||||
@Nonnull
|
||||
@Override
|
||||
public ByteBuf getPayload() {
|
||||
return payload.getBody();
|
||||
}
|
||||
};
|
||||
} else {
|
||||
wrap.message = new EncodedMessage() {
|
||||
@Nonnull
|
||||
@Override
|
||||
public ByteBuf getPayload() {
|
||||
return payload.getBody();
|
||||
}
|
||||
};
|
||||
}
|
||||
return wrap;
|
||||
}
|
||||
|
||||
@Nonnull
|
||||
@Override
|
||||
public String getTopic() {
|
||||
return topic;
|
||||
}
|
||||
|
||||
@Nonnull
|
||||
@Override
|
||||
public EncodedMessage getMessage() {
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,73 +0,0 @@
|
|||
package org.jetlinks.community.gateway.rule;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import lombok.Getter;
|
||||
import lombok.Setter;
|
||||
import org.jetlinks.community.gateway.TopicMessage;
|
||||
import org.jetlinks.rule.engine.api.RuleData;
|
||||
import org.jetlinks.rule.engine.api.RuleDataCodec;
|
||||
import org.jetlinks.rule.engine.api.RuleDataCodecs;
|
||||
import org.jetlinks.rule.engine.executor.PayloadType;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
public class TopicMessageCodec implements RuleDataCodec<TopicMessage> {
|
||||
|
||||
private static final TopicMessageCodec INSTANCE = new TopicMessageCodec();
|
||||
|
||||
static {
|
||||
RuleDataCodecs.register(TopicMessage.class, INSTANCE);
|
||||
}
|
||||
|
||||
public static void register() {
|
||||
}
|
||||
|
||||
|
||||
public static TopicMessageCodec getInstance() {
|
||||
return INSTANCE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, Object> encode(TopicMessage data, Feature... features) {
|
||||
|
||||
ByteBuf payload = data.getMessage().getPayload();
|
||||
PayloadType payloadType = PayloadType.valueOf(data.getMessage().getPayloadType().name());
|
||||
|
||||
Map<String, Object> map = new HashMap<>();
|
||||
map.put("topic", data.getTopic());
|
||||
map.put("message", payloadType.read(payload));
|
||||
return map;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<TopicMessage> decode(RuleData data, Feature... features) {
|
||||
|
||||
return Mono.fromSupplier(() -> Feature.find(TopicFeature.class, features)
|
||||
.map(TopicFeature::getTopics)
|
||||
.orElseThrow(() -> new UnsupportedOperationException("topics not found")))
|
||||
.flatMapMany(Flux::just)
|
||||
.flatMap(topic -> data
|
||||
.dataToMap()
|
||||
.map(map -> TopicMessage.of(topic, map)));
|
||||
}
|
||||
|
||||
|
||||
public static TopicFeature feature(String... topics) {
|
||||
return new TopicFeature(topics);
|
||||
}
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
public static class TopicFeature implements RuleDataCodec.Feature {
|
||||
|
||||
private String[] topics;
|
||||
|
||||
public TopicFeature(String... topics) {
|
||||
this.topics = topics;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -3,8 +3,6 @@ package org.jetlinks.community.gateway.spring;
|
|||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.hswebframework.web.bean.FastBeanCopier;
|
||||
import org.hswebframework.web.proxy.Proxy;
|
||||
import org.jetlinks.community.gateway.TopicMessage;
|
||||
import org.jetlinks.community.gateway.TopicMessageWrap;
|
||||
import org.jetlinks.core.NativePayload;
|
||||
import org.jetlinks.core.Payload;
|
||||
import org.jetlinks.core.codec.Codecs;
|
||||
|
|
@ -17,7 +15,6 @@ import reactor.core.publisher.Mono;
|
|||
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.StringJoiner;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
@Slf4j
|
||||
|
|
@ -80,10 +77,6 @@ class ProxyMessageListener implements MessageListener {
|
|||
return message;
|
||||
}
|
||||
try {
|
||||
if (paramType.equals(TopicMessage.class)) {
|
||||
log.warn("TopicMessage已弃用,请替换为TopicPayload! {}", method);
|
||||
return TopicMessageWrap.wrap(message);
|
||||
}
|
||||
Payload payload = message.getPayload();
|
||||
Object decodedPayload;
|
||||
if (payload instanceof NativePayload) {
|
||||
|
|
|
|||
|
|
@ -1,28 +0,0 @@
|
|||
package org.jetlinks.community.network.manager.web.response;
|
||||
|
||||
import lombok.Getter;
|
||||
import lombok.Setter;
|
||||
import org.jetlinks.community.gateway.TopicMessage;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
public class TopicMessageResponse {
|
||||
|
||||
private String topic;
|
||||
|
||||
private Object message;
|
||||
|
||||
public static TopicMessageResponse of(TopicMessage topicMessage) {
|
||||
TopicMessageResponse response = new TopicMessageResponse();
|
||||
response.setTopic(topicMessage.getTopic());
|
||||
response.setMessage(topicMessage.getMessage().getPayload().toString(StandardCharsets.UTF_8));
|
||||
return response;
|
||||
}
|
||||
|
||||
public TopicMessage toTopicMessage() {
|
||||
return TopicMessage.of(this.topic, this.message);
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue