diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionManagerImpl.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InMemorySessionManager.java similarity index 88% rename from weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionManagerImpl.java rename to weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InMemorySessionManager.java index e6de713d1..4d4d4c00b 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionManagerImpl.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InMemorySessionManager.java @@ -8,9 +8,9 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; -public class SessionManagerImpl implements WxSessionManager, InternalSessionManager { +public class InMemorySessionManager implements WxSessionManager, InternalSessionManager { - protected final Logger log = LoggerFactory.getLogger(SessionManagerImpl.class); + protected final Logger log = LoggerFactory.getLogger(InMemorySessionManager.class); protected static final StringManager sm = StringManager.getManager(Constants.Package); @@ -105,6 +105,11 @@ public class SessionManagerImpl implements WxSessionManager, InternalSessionMana */ protected int processExpiresFrequency = 6; + /** + * background processor delay in seconds + */ + protected int backgroundProcessorDelay = 10; + /** * 后台清理线程是否已经开启 */ @@ -204,7 +209,7 @@ public class SessionManagerImpl implements WxSessionManager, InternalSessionMana while (true) { try { // 每秒清理一次 - Thread.sleep(1000l); + Thread.sleep(backgroundProcessorDelay * 1000l); backgroundProcess(); } catch (InterruptedException e) { log.error("SessionManagerImpl.backgroundProcess error", e); @@ -225,6 +230,7 @@ public class SessionManagerImpl implements WxSessionManager, InternalSessionMana } } } + } /** @@ -269,6 +275,34 @@ public class SessionManagerImpl implements WxSessionManager, InternalSessionMana } + @Override + public void setMaxInactiveInterval(int interval) { + + this.maxInactiveInterval = interval; + + } + + /** + * Set the manager checks frequency. + * + * @param processExpiresFrequency the new manager checks frequency + */ + @Override + public void setProcessExpiresFrequency(int processExpiresFrequency) { + + if (processExpiresFrequency <= 0) { + return; + } + + this.processExpiresFrequency = processExpiresFrequency; + + } + + @Override + public void setBackgroundProcessorDelay(int backgroundProcessorDelay) { + this.backgroundProcessorDelay = backgroundProcessorDelay; + } + /** * Return the descriptive short name of this Manager implementation. */ diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionFacade.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionFacade.java index 458b40c18..d7c0bd62c 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionFacade.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionFacade.java @@ -10,7 +10,7 @@ public class InternalSessionFacade implements WxSession { private WxSession session = null; public InternalSessionFacade(WxSession session) { - session = session; + this.session = session; } @Override diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionManager.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionManager.java index 7699717a4..9dea5b715 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionManager.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionManager.java @@ -60,4 +60,15 @@ public interface InternalSessionManager { */ public void backgroundProcess(); + /** + * Set the default maximum inactive interval (in seconds) + * for Sessions created by this Manager. + * + * @param interval The new default value + */ + void setMaxInactiveInterval(int interval); + + void setProcessExpiresFrequency(int processExpiresFrequency); + + void setBackgroundProcessorDelay(int backgroundProcessorDelay); } diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionImpl.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionImpl.java index a899190d5..96cc470ba 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionImpl.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/SessionImpl.java @@ -308,4 +308,40 @@ public class SessionImpl implements WxSession, InternalSession { } + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof SessionImpl)) return false; + + SessionImpl session = (SessionImpl) o; + + if (creationTime != session.creationTime) return false; + if (expiring != session.expiring) return false; + if (isValid != session.isValid) return false; + if (maxInactiveInterval != session.maxInactiveInterval) return false; + if (thisAccessedTime != session.thisAccessedTime) return false; + if (!accessCount.equals(session.accessCount)) return false; + if (!attributes.equals(session.attributes)) return false; + if (!facade.equals(session.facade)) return false; + if (!id.equals(session.id)) return false; + if (!manager.equals(session.manager)) return false; + + return true; + } + + @Override + public int hashCode() { + int result = attributes.hashCode(); + result = 31 * result + id.hashCode(); + result = 31 * result + (isValid ? 1 : 0); + result = 31 * result + (expiring ? 1 : 0); + result = 31 * result + manager.hashCode(); + result = 31 * result + (int) (creationTime ^ (creationTime >>> 32)); + result = 31 * result + (int) (thisAccessedTime ^ (thisAccessedTime >>> 32)); + result = 31 * result + maxInactiveInterval; + result = 31 * result + facade.hashCode(); + result = 31 * result + accessCount.hashCode(); + return result; + } + } diff --git a/weixin-java-common/src/test/java/me/chanjar/weixin/common/session/TestSession.java b/weixin-java-common/src/test/java/me/chanjar/weixin/common/session/TestSession.java new file mode 100644 index 000000000..20e2dd884 --- /dev/null +++ b/weixin-java-common/src/test/java/me/chanjar/weixin/common/session/TestSession.java @@ -0,0 +1,68 @@ +package me.chanjar.weixin.common.session; + +import org.testng.Assert; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Test +public class TestSession { + + @DataProvider + public Object[][] getSessionManager() { + return new Object[][] { + new Object[] { new InMemorySessionManager() } + }; + } + + + @Test(dataProvider = "getSessionManager", expectedExceptions = IllegalStateException.class) + public void testInvalidate(WxSessionManager sessionManager) { + WxSession session = sessionManager.getSession("abc"); + session.invalidate(); + session.getAttributeNames(); + } + + @Test(dataProvider = "getSessionManager") + public void testInvalidate2(InternalSessionManager sessionManager) { + Assert.assertEquals(sessionManager.getActiveSessions(), 0); + WxSession session = ((WxSessionManager) sessionManager).getSession("abc"); + Assert.assertEquals(sessionManager.getActiveSessions(), 1); + session.invalidate(); + Assert.assertEquals(sessionManager.getActiveSessions(), 0); + } + + @Test(dataProvider = "getSessionManager") + public void testGetSession(WxSessionManager sessionManager) { + WxSession session1 = sessionManager.getSession("abc"); + WxSession session2 = sessionManager.getSession("abc"); + Assert.assertTrue(session1.equals(session2)); + + WxSession abc1 = sessionManager.getSession("abc1"); + Assert.assertFalse(session1.equals(abc1)); + + WxSession abc1b = sessionManager.getSession("abc1", false); + Assert.assertTrue(abc1.equals(abc1b)); + + WxSession def = sessionManager.getSession("def", false); + Assert.assertNull(def); + } + + @Test(dataProvider = "getSessionManager") + public void testBackgroundProcess(WxSessionManager sessionManager) throws InterruptedException { + + InternalSessionManager ism = (InternalSessionManager) sessionManager; + ism.setMaxInactiveInterval(1); + ism.setProcessExpiresFrequency(1); + ism.setBackgroundProcessorDelay(1); + + Assert.assertEquals(ism.getActiveSessions(), 0); + + InternalSession abc = ism.createSession("abc"); + abc.endAccess(); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + + } + +} diff --git a/weixin-java-common/src/test/resources/testng.xml b/weixin-java-common/src/test/resources/testng.xml index 984f0ae8e..ed4bf6ada 100644 --- a/weixin-java-common/src/test/resources/testng.xml +++ b/weixin-java-common/src/test/resources/testng.xml @@ -8,6 +8,7 @@ + diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageHandler.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageHandler.java index e4e26eb21..3b838781c 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageHandler.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageHandler.java @@ -1,5 +1,6 @@ package me.chanjar.weixin.cp.api; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import me.chanjar.weixin.cp.bean.WxCpXmlOutMessage; @@ -7,18 +8,21 @@ import java.util.Map; /** * 处理微信推送消息的处理器接口 - * @author Daniel Qian * + * @author Daniel Qian */ public interface WxCpMessageHandler { /** - * * @param wxMessage - * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 + * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 * @param wxCpService + * @param sessionManager * @return xml格式的消息，如果在异步规则里处理的话，可以返回null */ - public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService); - + public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, + Map context, + WxCpService wxCpService, + WxSessionManager sessionManager); + } diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageInterceptor.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageInterceptor.java index 5cb478e3d..92fa41f1b 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageInterceptor.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageInterceptor.java @@ -1,23 +1,29 @@ package me.chanjar.weixin.cp.api; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import java.util.Map; /** * 微信消息拦截器，可以用来做验证 - * @author Daniel Qian * + * @author Daniel Qian */ public interface WxCpMessageInterceptor { /** * 拦截微信消息 + * * @param wxMessage - * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 + * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 * @param wxCpService - * @return true代表OK，false代表不OK + * @param sessionManager + * @return true代表OK，false代表不OK */ - public boolean intercept(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService); - + public boolean intercept(WxCpXmlMessage wxMessage, + Map context, + WxCpService wxCpService, + WxSessionManager sessionManager); + } diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java index 5b9cd5746..cf831cf05 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java @@ -1,16 +1,24 @@ package me.chanjar.weixin.cp.api; +import me.chanjar.weixin.common.session.InternalSession; +import me.chanjar.weixin.common.session.WxSession; +import me.chanjar.weixin.common.session.InMemorySessionManager; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.WxMessageDuplicateChecker; import me.chanjar.weixin.common.util.WxMessageInMemoryDuplicateChecker; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import me.chanjar.weixin.cp.bean.WxCpXmlOutMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.regex.Pattern; /** @@ -43,6 +51,8 @@ import java.util.regex.Pattern; */ public class WxCpMessageRouter { + protected final Logger log = LoggerFactory.getLogger(WxCpMessageRouter.class); + private static final int DEFAULT_THREAD_POOL_SIZE = 100; private final List rules = new ArrayList(); @@ -51,16 +61,22 @@ public class WxCpMessageRouter { private ExecutorService executorService; - private WxMessageDuplicateChecker wxMessageDuplicateChecker; + private WxMessageDuplicateChecker messageDuplicateChecker; + + private WxSessionManager sessionManager; public WxCpMessageRouter(WxCpService wxCpService) { this.wxCpService = wxCpService; this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE); - this.wxMessageDuplicateChecker = new WxMessageInMemoryDuplicateChecker(); + this.messageDuplicateChecker = new WxMessageInMemoryDuplicateChecker(); + this.sessionManager = new InMemorySessionManager(); } /** - * 设置自定义的ExecutorService + * 
+   * 设置自定义的 {@link ExecutorService}
+   * 如果不调用该方法，默认使用 Executors.newFixedThreadPool(100)
+   *
* @param executorService */ public void setExecutorService(ExecutorService executorService) { @@ -68,11 +84,25 @@ public class WxCpMessageRouter { } /** - * 设置自定义的WxMsgIdDuplicateChecker - * @param wxMessageDuplicateChecker + * 
+   * 设置自定义的 {@link me.chanjar.weixin.common.util.WxMessageDuplicateChecker}
+   * 如果不调用该方法，默认使用 {@link me.chanjar.weixin.common.util.WxMessageInMemoryDuplicateChecker}
+   *
+ * @param messageDuplicateChecker */ - public void setWxMessageDuplicateChecker(WxMessageDuplicateChecker wxMessageDuplicateChecker) { - this.wxMessageDuplicateChecker = wxMessageDuplicateChecker; + public void setMessageDuplicateChecker(WxMessageDuplicateChecker messageDuplicateChecker) { + this.messageDuplicateChecker = messageDuplicateChecker; + } + + /** + * 
+   * 设置自定义的{@link me.chanjar.weixin.common.session.WxSessionManager}
+   * 如果不调用该方法，默认使用 {@linke SessionManagerImpl}
+   *
+ * @param sessionManager + */ + public void setSessionManager(WxSessionManager sessionManager) { + this.sessionManager = sessionManager; } /** @@ -80,7 +110,7 @@ public class WxCpMessageRouter { * @return */ public Rule rule() { - return new Rule(this, wxCpService); + return new Rule(this, wxCpService, sessionManager); } /** @@ -88,7 +118,7 @@ public class WxCpMessageRouter { * @param wxMessage */ public WxCpXmlOutMessage route(final WxCpXmlMessage wxMessage) { - if (wxMessageDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { + if (messageDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { // 如果是重复消息，那么就不做处理 return null; } @@ -109,27 +139,66 @@ public class WxCpMessageRouter { } WxCpXmlOutMessage res = null; + final List futures = new ArrayList(); for (final Rule rule : matchRules) { // 返回最后一个非异步的rule的执行结果 if(rule.async) { - executorService.submit(new Runnable() { - public void run() { - rule.service(wxMessage); - } - }); + futures.add( + executorService.submit(new Runnable() { + public void run() { + rule.service(wxMessage); + } + }) + ); } else { res = rule.service(wxMessage); } } + + // 告诉session，它已经用不着了 + if (futures.size() > 0) { + executorService.submit(new Runnable() { + @Override + public void run() { + for (Future future : futures) { + try { + future.get(); + } catch (InterruptedException e) { + log.error("Error happened when wait task finish", e); + } catch (ExecutionException e) { + log.error("Error happened when wait task finish", e); + } + } + // 在这里session再也不会被使用了 + sessionEndAccess(wxMessage); + } + }); + } else { + // 在这里session再也不会被使用了 + sessionEndAccess(wxMessage); + } return res; } + /** + * 对session的访问结束 + * @param wxMessage + */ + protected void sessionEndAccess(WxCpXmlMessage wxMessage) { + WxSession session = sessionManager.getSession(wxMessage.getFromUserName(), false); + if (session != null) { + ((InternalSession) session).endAccess(); + } + } + public static class Rule { private final WxCpMessageRouter routerBuilder; private final WxCpService wxCpService; + private final WxSessionManager sessionManager; + private boolean async = true; private String fromUser; @@ -152,9 +221,10 @@ public class WxCpMessageRouter { private List interceptors = new ArrayList(); - protected Rule(WxCpMessageRouter routerBuilder, WxCpService wxCpService) { + protected Rule(WxCpMessageRouter routerBuilder, WxCpService wxCpService, WxSessionManager sessionManager) { this.routerBuilder = routerBuilder; this.wxCpService = wxCpService; + this.sessionManager = sessionManager; } /** @@ -332,7 +402,7 @@ public class WxCpMessageRouter { Map context = new HashMap(); // 如果拦截器不通过 for (WxCpMessageInterceptor interceptor : this.interceptors) { - if (!interceptor.intercept(wxMessage, context, wxCpService)) { + if (!interceptor.intercept(wxMessage, context, wxCpService, sessionManager)) { return null; } } @@ -341,7 +411,7 @@ public class WxCpMessageRouter { WxCpXmlOutMessage res = null; for (WxCpMessageHandler handler : this.handlers) { // 返回最后handler的结果 - res = handler.handle(wxMessage, context, wxCpService); + res = handler.handle(wxMessage, context, wxCpService, sessionManager); } return res; } diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java index 29ec81771..e0b624bb2 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java @@ -392,7 +392,7 @@ public interface WxCpService { /** * 
    * 设置WxSessionManager，只有当需要使用个性化的WxSessionManager的时候才需要调用此方法，
-   * WxCpService默认使用的是{@link me.chanjar.weixin.common.session.SessionManagerImpl}
+   * WxCpService默认使用的是{@link me.chanjar.weixin.common.session.InMemorySessionManager}
    *
* @param sessionManager */ diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java index 473050e6a..1afcd7b30 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java @@ -12,7 +12,7 @@ import me.chanjar.weixin.common.bean.WxMenu; import me.chanjar.weixin.common.bean.result.WxError; import me.chanjar.weixin.common.bean.result.WxMediaUploadResult; import me.chanjar.weixin.common.exception.WxErrorException; -import me.chanjar.weixin.common.session.SessionManagerImpl; +import me.chanjar.weixin.common.session.InMemorySessionManager; import me.chanjar.weixin.common.session.WxSession; import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.StringUtils; @@ -46,7 +46,6 @@ import java.io.InputStream; import java.io.StringReader; import java.util.List; import java.util.UUID; -import java.util.concurrent.atomic.AtomicBoolean; public class WxCpServiceImpl implements WxCpService { @@ -67,7 +66,7 @@ public class WxCpServiceImpl implements WxCpService { private int maxRetryTimes = 5; - protected WxSessionManager sessionManager = new SessionManagerImpl(); + protected WxSessionManager sessionManager = new InMemorySessionManager(); public boolean checkSignature(String msgSignature, String timestamp, String nonce, String data) { try { diff --git a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpMessageRouterTest.java b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpMessageRouterTest.java index b3dc6c30a..0053a86bf 100644 --- a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpMessageRouterTest.java +++ b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpMessageRouterTest.java @@ -1,6 +1,7 @@ package me.chanjar.weixin.cp.api; import me.chanjar.weixin.common.api.WxConsts; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import me.chanjar.weixin.cp.bean.WxCpXmlOutMessage; import org.testng.Assert; @@ -67,7 +68,8 @@ public class WxCpMessageRouterTest { final WxCpMessageRouter router = new WxCpMessageRouter(null); router.rule().handler(new WxCpMessageHandler() { @Override - public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService) { + public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService, + WxSessionManager sessionManager) { return null; } }).end(); @@ -149,7 +151,8 @@ public class WxCpMessageRouterTest { } @Override - public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService) { + public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService, + WxSessionManager sessionManager) { sb.append(this.echoStr).append(','); return null; } diff --git a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpDemoServer.java b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpDemoServer.java index 817861003..0dadee91d 100644 --- a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpDemoServer.java +++ b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpDemoServer.java @@ -1,5 +1,6 @@ package me.chanjar.weixin.cp.demo; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.cp.api.*; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import me.chanjar.weixin.cp.bean.WxCpXmlOutMessage; @@ -46,7 +47,7 @@ public class WxCpDemoServer { WxCpMessageHandler handler = new WxCpMessageHandler() { @Override public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, - WxCpService wxCpService) { + WxCpService wxCpService, WxSessionManager sessionManager) { WxCpXmlOutTextMessage m = WxCpXmlOutMessage .TEXT() .content("测试加密消息") @@ -60,7 +61,7 @@ public class WxCpDemoServer { WxCpMessageHandler oauth2handler = new WxCpMessageHandler() { @Override public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, - WxCpService wxCpService) { + WxCpService wxCpService, WxSessionManager sessionManager) { String href = "测试oauth2"; return WxCpXmlOutMessage diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageHandler.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageHandler.java index 8e6deeb6c..d7c30a31b 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageHandler.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageHandler.java @@ -1,5 +1,6 @@ package me.chanjar.weixin.mp.api; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.mp.bean.WxMpXmlMessage; import me.chanjar.weixin.mp.bean.WxMpXmlOutMessage; @@ -7,18 +8,21 @@ import java.util.Map; /** * 处理微信推送消息的处理器接口 - * @author chanjarster * + * @author Daniel Qian */ public interface WxMpMessageHandler { /** - * * @param wxMessage - * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 + * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 * @param wxMpService + * @param sessionManager * @return xml格式的消息，如果在异步规则里处理的话，可以返回null */ - public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService); - + public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, + Map context, + WxMpService wxMpService, + WxSessionManager sessionManager); + } diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageInterceptor.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageInterceptor.java index 2e3825153..2b68a6f20 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageInterceptor.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageInterceptor.java @@ -1,21 +1,29 @@ package me.chanjar.weixin.mp.api; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.mp.bean.WxMpXmlMessage; import java.util.Map; /** * 微信消息拦截器，可以用来做验证 + * + * @author Daniel Qian */ public interface WxMpMessageInterceptor { /** * 拦截微信消息 + * * @param wxMessage - * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 + * @param context 上下文，如果handler或interceptor之间有信息要传递，可以用这个 * @param wxMpService - * @return true代表OK，false代表不OK + * @param sessionManager + * @return true代表OK，false代表不OK */ - public boolean intercept(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService); - + public boolean intercept(WxMpXmlMessage wxMessage, + Map context, + WxMpService wxMpService, + WxSessionManager sessionManager); + } diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java index a46d5716f..c35373805 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java @@ -1,7 +1,7 @@ package me.chanjar.weixin.mp.api; import me.chanjar.weixin.common.session.InternalSession; -import me.chanjar.weixin.common.session.SessionManagerImpl; +import me.chanjar.weixin.common.session.InMemorySessionManager; import me.chanjar.weixin.common.session.WxSession; import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.WxMessageDuplicateChecker; @@ -59,15 +59,15 @@ public class WxMpMessageRouter { private ExecutorService executorService; - private WxMessageDuplicateChecker wxMessageDuplicateChecker; + private WxMessageDuplicateChecker messageDuplicateChecker; private WxSessionManager sessionManager; public WxMpMessageRouter(WxMpService wxMpService) { this.wxMpService = wxMpService; this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE); - this.wxMessageDuplicateChecker = new WxMessageInMemoryDuplicateChecker(); - this.sessionManager = new SessionManagerImpl(); + this.messageDuplicateChecker = new WxMessageInMemoryDuplicateChecker(); + this.sessionManager = new InMemorySessionManager(); } /** @@ -86,10 +86,10 @@ public class WxMpMessageRouter { * 设置自定义的 {@link me.chanjar.weixin.common.util.WxMessageDuplicateChecker} * 如果不调用该方法，默认使用 {@link me.chanjar.weixin.common.util.WxMessageInMemoryDuplicateChecker} * - * @param wxMessageDuplicateChecker + * @param messageDuplicateChecker */ - public void setWxMessageDuplicateChecker(WxMessageDuplicateChecker wxMessageDuplicateChecker) { - this.wxMessageDuplicateChecker = wxMessageDuplicateChecker; + public void setMessageDuplicateChecker(WxMessageDuplicateChecker messageDuplicateChecker) { + this.messageDuplicateChecker = messageDuplicateChecker; } /** @@ -108,7 +108,7 @@ public class WxMpMessageRouter { * @return */ public Rule rule() { - return new Rule(this, wxMpService); + return new Rule(this, wxMpService, sessionManager); } /** @@ -116,7 +116,7 @@ public class WxMpMessageRouter { * @param wxMessage */ public WxMpXmlOutMessage route(final WxMpXmlMessage wxMessage) { - if (wxMessageDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { + if (messageDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { // 如果是重复消息，那么就不做处理 return null; } @@ -195,6 +195,8 @@ public class WxMpMessageRouter { private final WxMpService wxMpService; + private final WxSessionManager sessionManager; + private boolean async = true; private String fromUser; @@ -214,10 +216,11 @@ public class WxMpMessageRouter { private List handlers = new ArrayList(); private List interceptors = new ArrayList(); - - protected Rule(WxMpMessageRouter routerBuilder, WxMpService wxMpService) { + + protected Rule(WxMpMessageRouter routerBuilder, WxMpService wxMpService, WxSessionManager sessionManager) { this.routerBuilder = routerBuilder; this.wxMpService = wxMpService; + this.sessionManager = sessionManager; } /** @@ -383,7 +386,7 @@ public class WxMpMessageRouter { Map context = new HashMap(); // 如果拦截器不通过 for (WxMpMessageInterceptor interceptor : this.interceptors) { - if (!interceptor.intercept(wxMessage, context, wxMpService)) { + if (!interceptor.intercept(wxMessage, context, wxMpService, sessionManager)) { return null; } } @@ -392,7 +395,7 @@ public class WxMpMessageRouter { WxMpXmlOutMessage res = null; for (WxMpMessageHandler handler : this.handlers) { // 返回最后handler的结果 - res = handler.handle(wxMessage, context, wxMpService); + res = handler.handle(wxMessage, context, wxMpService, sessionManager); } return res; } diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java index f3c96f5b8..21a8b94c5 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java @@ -10,8 +10,7 @@ import me.chanjar.weixin.common.bean.WxMenu; import me.chanjar.weixin.common.bean.result.WxError; import me.chanjar.weixin.common.bean.result.WxMediaUploadResult; import me.chanjar.weixin.common.exception.WxErrorException; -import me.chanjar.weixin.common.session.SessionManagerImpl; -import me.chanjar.weixin.common.session.WxSession; +import me.chanjar.weixin.common.session.InMemorySessionManager; import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.StringUtils; import me.chanjar.weixin.common.util.crypto.SHA1; @@ -69,7 +68,7 @@ public class WxMpServiceImpl implements WxMpService { private int maxRetryTimes = 5; - protected WxSessionManager sessionManager = new SessionManagerImpl(); + protected WxSessionManager sessionManager = new InMemorySessionManager(); public boolean checkSignature(String timestamp, String nonce, String signature) { try { diff --git a/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpMessageRouterTest.java b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpMessageRouterTest.java index 3f7e841ee..15506c857 100644 --- a/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpMessageRouterTest.java +++ b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpMessageRouterTest.java @@ -1,6 +1,7 @@ package me.chanjar.weixin.mp.api; import me.chanjar.weixin.common.api.WxConsts; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.mp.bean.WxMpXmlMessage; import me.chanjar.weixin.mp.bean.WxMpXmlOutMessage; import org.testng.Assert; @@ -67,7 +68,8 @@ public class WxMpMessageRouterTest { final WxMpMessageRouter router = new WxMpMessageRouter(null); router.rule().handler(new WxMpMessageHandler() { @Override - public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService) { + public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService, + WxSessionManager sessionManager) { return null; } }).end(); @@ -149,7 +151,8 @@ public class WxMpMessageRouterTest { } @Override - public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService) { + public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService, + WxSessionManager sessionManager) { sb.append(this.echoStr).append(','); return null; } diff --git a/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/demo/WxMpDemoServer.java b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/demo/WxMpDemoServer.java index c58671cf3..8e44d2db9 100644 --- a/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/demo/WxMpDemoServer.java +++ b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/demo/WxMpDemoServer.java @@ -3,6 +3,7 @@ package me.chanjar.weixin.mp.demo; import me.chanjar.weixin.common.api.WxConsts; import me.chanjar.weixin.common.bean.result.WxMediaUploadResult; import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.mp.api.*; import me.chanjar.weixin.mp.bean.WxMpXmlMessage; import me.chanjar.weixin.mp.bean.WxMpXmlOutImageMessage; @@ -51,7 +52,7 @@ public class WxMpDemoServer { WxMpMessageHandler textHandler = new WxMpMessageHandler() { @Override public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, - WxMpService wxMpService) { + WxMpService wxMpService, WxSessionManager sessionManager) { WxMpXmlOutTextMessage m = WxMpXmlOutMessage.TEXT().content("测试加密消息").fromUser(wxMessage.getToUserName()) .toUser(wxMessage.getFromUserName()).build(); @@ -62,7 +63,7 @@ public class WxMpDemoServer { WxMpMessageHandler imageHandler = new WxMpMessageHandler() { @Override public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, - WxMpService wxMpService) { + WxMpService wxMpService, WxSessionManager sessionManager) { try { WxMediaUploadResult wxMediaUploadResult = wxMpService .mediaUpload(WxConsts.MEDIA_IMAGE, WxConsts.FILE_JPG, ClassLoader.getSystemResourceAsStream("mm.jpeg")); @@ -86,7 +87,7 @@ public class WxMpDemoServer { WxMpMessageHandler oauth2handler = new WxMpMessageHandler() { @Override public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, - WxMpService wxMpService) { + WxMpService wxMpService, WxSessionManager sessionManager) { String href = "测试oauth2"; return WxMpXmlOutMessage