feat(all): 修改mq发送消息根据租户切换的问题

blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/config/CustomMessageConverter.java

@@ -6,14 +6,25 @@ import org.springblade.core.secure.utils.AuthUtil;
 import org.springblade.core.tool.utils.ThreadLocalUtil;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.AllowedListDeserializingMessageConverter;
 import org.springframework.amqp.support.converter.MessageConversionException;
-import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.http.HttpHeaders;
 import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;
 import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
 import org.springframework.web.context.request.RequestContextHolder;
 import org.springframework.web.context.request.ServletRequestAttributes;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
 import java.util.Map;
 
 /**
@@ -21,9 +32,74 @@ import java.util.Map;
  * @create 2024-05-08
  */
 @Component
-public class CustomMessageConverter implements MessageConverter {
+public class CustomMessageConverter extends AllowedListDeserializingMessageConverter implements BeanClassLoaderAware {
+
+	private volatile String defaultCharset = "UTF-8";
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+	private String codebaseUrl;
+
+	@Override
+	public Object fromMessage(Message message) throws MessageConversionException {
+		Object content = null;
+		MessageProperties properties = message.getMessageProperties();
+		if (properties != null) {
+			String contentType = properties.getContentType();
+			if (contentType != null && contentType.startsWith("text")) {
+				String encoding = properties.getContentEncoding();
+				if (encoding == null) {
+					encoding = "UTF-8";
+				}
+
+				try {
+					content = new String(message.getBody(), encoding);
+				} catch (UnsupportedEncodingException var8) {
+					throw new MessageConversionException("failed to convert text-based Message content", var8);
+				}
+			} else if (contentType != null && contentType.equals("application/x-java-serialized-object")) {
+				try {
+					content = SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
+				} catch (IllegalArgumentException | IllegalStateException | IOException var7) {
+					throw new MessageConversionException("failed to convert serialized Message content", var7);
+				}
+			}
+		}
+		Map<String, Object> headers = properties.getHeaders();
+		ThreadLocalUtil.put("bladeContext", headers);
+		Object bladeUser = headers.get("bladeUser");
+		MockHttpServletRequest request = new MockHttpServletRequest();
+		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
+		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
+		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
+		if (content == null) {
+			content = message.getBody();
+		}
+		return content;
+	}
+
 	@Override
-	public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
+	protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		byte[] bytes = null;
+		if (object instanceof byte[]) {
+			bytes = (byte[]) object;
+			messageProperties.setContentType("application/octet-stream");
+		} else if (object instanceof String) {
+			try {
+				bytes = ((String) object).getBytes(this.defaultCharset);
+			} catch (UnsupportedEncodingException var6) {
+				throw new MessageConversionException("failed to convert to Message content", var6);
+			}
+
+			messageProperties.setContentType("text/plain");
+			messageProperties.setContentEncoding(this.defaultCharset);
+		} else if (object instanceof Serializable) {
+			try {
+				bytes = SerializationUtils.serialize(object);
+			} catch (IllegalArgumentException var5) {
+				throw new MessageConversionException("failed to convert to serialized Message content", var5);
+			}
+
+			messageProperties.setContentType("application/x-java-serialized-object");
+		}
 		HttpHeaders headers = (HttpHeaders) ThreadLocalUtil.get("bladeContext");
 		if (headers != null && !headers.isEmpty()) {
 			headers.forEach((key, values) -> {
@@ -39,19 +115,28 @@ public class CustomMessageConverter implements MessageConverter {
 		bladeUser.setAccount(user.getAccount());
 		bladeUser.setRoleId(user.getRoleId());
 		messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser));
-		return new Message(o.toString().getBytes(), messageProperties);
+
+		if (bytes != null) {
+			messageProperties.setContentLength((long) bytes.length);
+			return new Message(bytes, messageProperties);
+		} else {
+			throw new IllegalArgumentException(this.getClass().getSimpleName() + " only supports String, byte[] and Serializable payloads, received: " + object.getClass().getName());
+		}
 	}
 
 	@Override
-	public Object fromMessage(Message message) throws MessageConversionException {
-		MessageProperties messageProperties = message.getMessageProperties();
-		Map<String, Object> headers = messageProperties.getHeaders();
-		ThreadLocalUtil.put("bladeContext", headers);
-		Object bladeUser = headers.get("bladeUser");
-		MockHttpServletRequest request = new MockHttpServletRequest();
-		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
-		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
-		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
-		return new String(message.getBody());
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = beanClassLoader;
+	}
+
+	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
+		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl) {
+			@Override
+			protected Class<?> resolveClass(ObjectStreamClass classDesc) throws IOException, ClassNotFoundException {
+				Class<?> clazz = super.resolveClass(classDesc);
+				CustomMessageConverter.this.checkAllowedList(clazz);
+				return clazz;
+			}
+		};
 	}
 }