From be5f644173e57b1a7a1c9b188c721de0f859d8eb Mon Sep 17 00:00:00 2001 From: zhaoqiaobo Date: Wed, 8 May 2024 21:16:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(all):=20=E4=BF=AE=E6=94=B9mq=E5=8F=91?= =?UTF-8?q?=E9=80=81=E6=B6=88=E6=81=AF=E6=A0=B9=E6=8D=AE=E7=A7=9F=E6=88=B7?= =?UTF-8?q?=E5=88=87=E6=8D=A2=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/CustomMessageConverter.java | 2 +- .../base/config/CustomMessageConverter.java | 119 ++++++++++++--- .../jinpai/aspect/AnnotationAspect.java | 45 ++++++ .../jinpai/config/CustomMessageConverter.java | 142 ++++++++++++++++++ .../jinpai/config/RabbitConfig.java | 17 +++ .../olo/aspect/AnnotationAspect.java | 12 ++ .../factorydata/olo/config/RabbitConfig.java | 5 +- .../suofeiya/aspect/AnnotationAspect.java | 45 ++++++ .../config/CustomMessageConverter.java | 142 ++++++++++++++++++ .../suofeiya/config/RabbitConfig.java | 17 +++ .../zbom/aspect/AnnotationAspect.java | 45 ++++++ .../zbom/config/CustomMessageConverter.java | 142 ++++++++++++++++++ .../factorydata/zbom/config/RabbitConfig.java | 17 +++ .../config/CustomMessageConverter.java | 2 +- .../config/CustomMessageConverter.java | 2 +- .../config/CustomMessageConverter.java | 2 +- 16 files changed, 728 insertions(+), 28 deletions(-) create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/aspect/AnnotationAspect.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/CustomMessageConverter.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/RabbitConfig.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/aspect/AnnotationAspect.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/CustomMessageConverter.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/RabbitConfig.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/aspect/AnnotationAspect.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/CustomMessageConverter.java create mode 100644 blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/RabbitConfig.java diff --git a/blade-service/logpm-distribution/src/main/java/com/logpm/distribution/config/CustomMessageConverter.java b/blade-service/logpm-distribution/src/main/java/com/logpm/distribution/config/CustomMessageConverter.java index ffa9e4a7e..3f8703627 100644 --- a/blade-service/logpm-distribution/src/main/java/com/logpm/distribution/config/CustomMessageConverter.java +++ b/blade-service/logpm-distribution/src/main/java/com/logpm/distribution/config/CustomMessageConverter.java @@ -1,4 +1,4 @@ -package com.logpm.factorydata.olo.config; +package com.logpm.distribution.config; import cn.hutool.json.JSONUtil; import org.springblade.core.secure.BladeUser; diff --git a/blade-service/logpm-factory-data/logpm-factory-data-base/src/main/java/com/logpm/factorydata/base/config/CustomMessageConverter.java b/blade-service/logpm-factory-data/logpm-factory-data-base/src/main/java/com/logpm/factorydata/base/config/CustomMessageConverter.java index 94d1e454e..71b65cfac 100644 --- a/blade-service/logpm-factory-data/logpm-factory-data-base/src/main/java/com/logpm/factorydata/base/config/CustomMessageConverter.java +++ b/blade-service/logpm-factory-data/logpm-factory-data-base/src/main/java/com/logpm/factorydata/base/config/CustomMessageConverter.java @@ -6,24 +6,100 @@ 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; /** * @author zhaoqiaobo * @create 2024-05-08 */ -public class CustomMessageConverter implements MessageConverter { - private String name; +@Component +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 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,27 +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); - } - @Override - public Object fromMessage(Message message) throws MessageConversionException { - MessageProperties messageProperties = message.getMessageProperties(); - Map 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()); + 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()); + } } - public String getName() { - return name; + @Override + public void setBeanClassLoader(ClassLoader classLoader) { + this.beanClassLoader = beanClassLoader; } - public void setName(String name) { - this.name = name; + 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; + } + }; } } diff --git a/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/aspect/AnnotationAspect.java b/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/aspect/AnnotationAspect.java new file mode 100644 index 000000000..2ee23520a --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/aspect/AnnotationAspect.java @@ -0,0 +1,45 @@ +package com.logpm.factorydata.jinpai.aspect; + +import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.springblade.core.secure.utils.AuthUtil; +import org.springframework.stereotype.Component; + +@Aspect +@Component +public class AnnotationAspect { + + /** + * 该方法为一个切面，用于对标记了@RabbitListener注解的方法进行增强，主要增加了日志记录和数据源上下文处理的功能。 + * + * @param joinPoint 切面的连接点，表示当前被拦截的方法。 + * @return 返回被拦截方法的执行结果。 + * @throws Throwable 如果执行过程中发生异常，则抛出。 + */ + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)") + public Object logAroundAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)") + public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/CustomMessageConverter.java b/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/CustomMessageConverter.java new file mode 100644 index 000000000..f50a2a6cc --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/CustomMessageConverter.java @@ -0,0 +1,142 @@ +package com.logpm.factorydata.jinpai.config; + +import cn.hutool.json.JSONUtil; +import org.springblade.core.secure.BladeUser; +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.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; + +/** + * @author zhaoqiaobo + * @create 2024-05-08 + */ +@Component +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 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 + 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) -> { + values.forEach((value) -> { + messageProperties.setHeader(key, new String[]{value}); + }); + }); + } + BladeUser user = AuthUtil.getUser(); + BladeUser bladeUser = new BladeUser(); + bladeUser.setTenantId(user.getTenantId()); + bladeUser.setUserId(user.getUserId()); + bladeUser.setAccount(user.getAccount()); + bladeUser.setRoleId(user.getRoleId()); + messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser)); + + 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 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; + } + }; + } +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/RabbitConfig.java b/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/RabbitConfig.java new file mode 100644 index 000000000..b1bfa1153 --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/RabbitConfig.java @@ -0,0 +1,17 @@ +package com.logpm.factorydata.jinpai.config; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RabbitConfig { + + @Bean + public RabbitTemplate rabbitTemplate(ConnectionFactory factory) { + RabbitTemplate template = new RabbitTemplate(factory); + template.setMessageConverter(new CustomMessageConverter()); + return template; + } +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/aspect/AnnotationAspect.java b/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/aspect/AnnotationAspect.java index b8ad44dc6..b5a9dc38d 100644 --- a/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/aspect/AnnotationAspect.java +++ b/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/aspect/AnnotationAspect.java @@ -30,4 +30,16 @@ public class AnnotationAspect { return result; } + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)") + public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + } diff --git a/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/config/RabbitConfig.java b/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/config/RabbitConfig.java index 46ef3cfc9..caf25248c 100644 --- a/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/config/RabbitConfig.java +++ b/blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/config/RabbitConfig.java @@ -1,6 +1,5 @@ package com.logpm.factorydata.olo.config; -import com.logpm.factorydata.olo.config.CustomMessageConverter; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.Bean; @@ -10,9 +9,9 @@ import org.springframework.context.annotation.Configuration; public class RabbitConfig { @Bean - public RabbitTemplate rabbitTemplate(ConnectionFactory factory, CustomMessageConverter customMessageConverter) { + public RabbitTemplate rabbitTemplate(ConnectionFactory factory, CustomMessageConverter converter) { RabbitTemplate template = new RabbitTemplate(factory); - template.setMessageConverter(customMessageConverter); + template.setMessageConverter(converter); return template; } } diff --git a/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/aspect/AnnotationAspect.java b/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/aspect/AnnotationAspect.java new file mode 100644 index 000000000..1c949f0fc --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/aspect/AnnotationAspect.java @@ -0,0 +1,45 @@ +package com.logpm.factorydata.suofeiya.aspect; + +import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.springblade.core.secure.utils.AuthUtil; +import org.springframework.stereotype.Component; + +@Aspect +@Component +public class AnnotationAspect { + + /** + * 该方法为一个切面，用于对标记了@RabbitListener注解的方法进行增强，主要增加了日志记录和数据源上下文处理的功能。 + * + * @param joinPoint 切面的连接点，表示当前被拦截的方法。 + * @return 返回被拦截方法的执行结果。 + * @throws Throwable 如果执行过程中发生异常，则抛出。 + */ + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)") + public Object logAroundAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)") + public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/CustomMessageConverter.java b/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/CustomMessageConverter.java new file mode 100644 index 000000000..acc4e05ce --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/CustomMessageConverter.java @@ -0,0 +1,142 @@ +package com.logpm.factorydata.suofeiya.config; + +import cn.hutool.json.JSONUtil; +import org.springblade.core.secure.BladeUser; +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.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; + +/** + * @author zhaoqiaobo + * @create 2024-05-08 + */ +@Component +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 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 + 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) -> { + values.forEach((value) -> { + messageProperties.setHeader(key, new String[]{value}); + }); + }); + } + BladeUser user = AuthUtil.getUser(); + BladeUser bladeUser = new BladeUser(); + bladeUser.setTenantId(user.getTenantId()); + bladeUser.setUserId(user.getUserId()); + bladeUser.setAccount(user.getAccount()); + bladeUser.setRoleId(user.getRoleId()); + messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser)); + + 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 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; + } + }; + } +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/RabbitConfig.java b/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/RabbitConfig.java new file mode 100644 index 000000000..194f0d2b2 --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/RabbitConfig.java @@ -0,0 +1,17 @@ +package com.logpm.factorydata.suofeiya.config; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RabbitConfig { + + @Bean + public RabbitTemplate rabbitTemplate(ConnectionFactory factory) { + RabbitTemplate template = new RabbitTemplate(factory); + template.setMessageConverter(new CustomMessageConverter()); + return template; + } +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/aspect/AnnotationAspect.java b/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/aspect/AnnotationAspect.java new file mode 100644 index 000000000..554dff7af --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/aspect/AnnotationAspect.java @@ -0,0 +1,45 @@ +package com.logpm.factorydata.zbom.aspect; + +import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.springblade.core.secure.utils.AuthUtil; +import org.springframework.stereotype.Component; + +@Aspect +@Component +public class AnnotationAspect { + + /** + * 该方法为一个切面，用于对标记了@RabbitListener注解的方法进行增强，主要增加了日志记录和数据源上下文处理的功能。 + * + * @param joinPoint 切面的连接点，表示当前被拦截的方法。 + * @return 返回被拦截方法的执行结果。 + * @throws Throwable 如果执行过程中发生异常，则抛出。 + */ + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)") + public Object logAroundAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + + @Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)") + public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable { + // 在方法执行前，获取当前租户ID并设置到数据源上下文中 + String tenantId = AuthUtil.getTenantId(); + DynamicDataSourceContextHolder.push(tenantId); + // 执行原方法 + Object result = joinPoint.proceed(); + // 在方法执行后，从数据源上下文中移除租户ID + DynamicDataSourceContextHolder.poll(); + return result; + } + +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/CustomMessageConverter.java b/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/CustomMessageConverter.java new file mode 100644 index 000000000..89e07ecef --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/CustomMessageConverter.java @@ -0,0 +1,142 @@ +package com.logpm.factorydata.zbom.config; + +import cn.hutool.json.JSONUtil; +import org.springblade.core.secure.BladeUser; +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.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; + +/** + * @author zhaoqiaobo + * @create 2024-05-08 + */ +@Component +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 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 + 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) -> { + values.forEach((value) -> { + messageProperties.setHeader(key, new String[]{value}); + }); + }); + } + BladeUser user = AuthUtil.getUser(); + BladeUser bladeUser = new BladeUser(); + bladeUser.setTenantId(user.getTenantId()); + bladeUser.setUserId(user.getUserId()); + bladeUser.setAccount(user.getAccount()); + bladeUser.setRoleId(user.getRoleId()); + messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser)); + + 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 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; + } + }; + } +} diff --git a/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/RabbitConfig.java b/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/RabbitConfig.java new file mode 100644 index 000000000..7b4e22d15 --- /dev/null +++ b/blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/RabbitConfig.java @@ -0,0 +1,17 @@ +package com.logpm.factorydata.zbom.config; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RabbitConfig { + + @Bean + public RabbitTemplate rabbitTemplate(ConnectionFactory factory) { + RabbitTemplate template = new RabbitTemplate(factory); + template.setMessageConverter(new CustomMessageConverter()); + return template; + } +} diff --git a/blade-service/logpm-factory/src/main/java/com/logpm/factory/config/CustomMessageConverter.java b/blade-service/logpm-factory/src/main/java/com/logpm/factory/config/CustomMessageConverter.java index ffa9e4a7e..deb695b4c 100644 --- a/blade-service/logpm-factory/src/main/java/com/logpm/factory/config/CustomMessageConverter.java +++ b/blade-service/logpm-factory/src/main/java/com/logpm/factory/config/CustomMessageConverter.java @@ -1,4 +1,4 @@ -package com.logpm.factorydata.olo.config; +package com.logpm.factory.config; import cn.hutool.json.JSONUtil; import org.springblade.core.secure.BladeUser; diff --git a/blade-service/logpm-trunkline/src/main/java/com/logpm/trunkline/config/CustomMessageConverter.java b/blade-service/logpm-trunkline/src/main/java/com/logpm/trunkline/config/CustomMessageConverter.java index ffa9e4a7e..420eaaaa9 100644 --- a/blade-service/logpm-trunkline/src/main/java/com/logpm/trunkline/config/CustomMessageConverter.java +++ b/blade-service/logpm-trunkline/src/main/java/com/logpm/trunkline/config/CustomMessageConverter.java @@ -1,4 +1,4 @@ -package com.logpm.factorydata.olo.config; +package com.logpm.trunkline.config; import cn.hutool.json.JSONUtil; import org.springblade.core.secure.BladeUser; diff --git a/blade-service/logpm-warehouse/src/main/java/com/logpm/warehouse/config/CustomMessageConverter.java b/blade-service/logpm-warehouse/src/main/java/com/logpm/warehouse/config/CustomMessageConverter.java index ffa9e4a7e..bdba5446b 100644 --- a/blade-service/logpm-warehouse/src/main/java/com/logpm/warehouse/config/CustomMessageConverter.java +++ b/blade-service/logpm-warehouse/src/main/java/com/logpm/warehouse/config/CustomMessageConverter.java @@ -1,4 +1,4 @@ -package com.logpm.factorydata.olo.config; +package com.logpm.warehouse.config; import cn.hutool.json.JSONUtil; import org.springblade.core.secure.BladeUser;