diff --git a/pom.xml b/pom.xml index 83d777a2..01cadc11 100644 --- a/pom.xml +++ b/pom.xml @@ -1,219 +1,232 @@ - 4.0.0 - me.chanjar - weixin-java-parent - 1.0.7-SNAPSHOT - pom - WeiXin Java Tools - Parent - 微信公众号、企业号上级POM - https://github.com/chanjarster/weixin-java-tools + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" + xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"> + 4.0.0 + me.chanjar + weixin-java-parent + 1.0.7-SNAPSHOT + pom + WeiXin Java Tools - Parent + 微信公众号、企业号上级POM + https://github.com/chanjarster/weixin-java-tools + + + + The Apache License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0.txt + + - - - The Apache License, Version 2.0 - http://www.apache.org/licenses/LICENSE-2.0.txt - - + + + Daniel Qian + chanjarster@gmail.com + + - - - Daniel Qian - chanjarster@gmail.com - - + + scm:git:https://github.com/chanjarster/weixin-java-tools.git + scm:git:git@github.com:chanjarster/weixin-java-tools.git + https://github.com/chanjarster/weixin-java-tools + - - scm:git:https://github.com/chanjarster/weixin-java-tools.git - scm:git:git@github.com:chanjarster/weixin-java-tools.git - https://github.com/chanjarster/weixin-java-tools - + + weixin-java-common + weixin-java-cp + weixin-java-mp + - - weixin-java-common - weixin-java-cp - weixin-java-mp - + + UTF-8 + true + true + 4.3.5 + 1.7.10 + 1.1.2 + - - UTF-8 - true - true - 4.3.5 - + + + org.slf4j + slf4j-api + ${slf4j.version} + + + ch.qos.logback + logback-classic + ${logback.version} + test + + + org.apache.httpcomponents + fluent-hc + ${httpclient.version} + + + org.apache.httpcomponents + httpmime + ${httpclient.version} + + + com.google.code.gson + gson + 2.2.2 + + + commons-codec + commons-codec + 1.9 + + + commons-io + commons-io + 2.4 + + + - - org.apache.httpcomponents - fluent-hc - ${httpclient.version} - - - org.apache.httpcomponents - httpmime - ${httpclient.version} - - - com.google.code.gson - gson - 2.2.2 - - - commons-codec - commons-codec - 1.9 - - - commons-io - commons-io - 2.4 - + + junit + junit + 4.11 + test + + + com.google.inject + guice + 3.0 + test + + + org.testng + testng + 6.8.7 + test + + + org.mockito + mockito-all + 1.9.5 + test + + + org.eclipse.jetty + jetty-server + 9.3.0.M0 + test + + + org.eclipse.jetty + jetty-servlet + 9.3.0.M0 + test + + - - - - junit - junit - 4.11 - test - - - com.google.inject - guice - 3.0 - test - - - org.testng - testng - 6.8.7 - test - - - org.mockito - mockito-all - 1.9.5 - test - - - org.eclipse.jetty - jetty-server - 9.3.0.M0 - test - - - org.eclipse.jetty - jetty-servlet - 9.3.0.M0 - test - - - - - - - ossrh - https://oss.sonatype.org/content/repositories/snapshots - - - ossrh - https://oss.sonatype.org/service/local/staging/deploy/maven2/ - - - - - - release - - - - org.apache.maven.plugins - maven-source-plugin - 2.2.1 - - - attach-sources - - jar-no-fork - - - - - - org.apache.maven.plugins - maven-javadoc-plugin - 2.9.1 - - - attach-javadocs - - jar - - - - - UTF-8 - zh_CN - - - - org.apache.maven.plugins - maven-gpg-plugin - 1.5 - - - sign-artifacts - verify - - sign - - - - - - - - - - - - - - org.apache.maven.plugins - maven-surefire-plugin - 2.17 - - - + + + ossrh + https://oss.sonatype.org/content/repositories/snapshots + + + ossrh + https://oss.sonatype.org/service/local/staging/deploy/maven2/ + + + + + release + - - org.sonatype.plugins - nexus-staging-maven-plugin - 1.6.3 - true - - ossrh - https://oss.sonatype.org/ - false - - - - org.apache.maven.plugins - maven-release-plugin - 2.5 - - true - false - release - deploy - - + + org.apache.maven.plugins + maven-source-plugin + 2.2.1 + + + attach-sources + + jar-no-fork + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + 2.9.1 + + + attach-javadocs + + jar + + + + + UTF-8 + zh_CN + + + + org.apache.maven.plugins + maven-gpg-plugin + 1.5 + + + sign-artifacts + verify + + sign + + + + - + + + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + 2.17 + + + + + + + org.sonatype.plugins + nexus-staging-maven-plugin + 1.6.3 + true + + ossrh + https://oss.sonatype.org/ + false + + + + org.apache.maven.plugins + maven-release-plugin + 2.5 + + true + false + release + deploy + + + + diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/Constants.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/Constants.java new file mode 100644 index 00000000..33518f17 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/Constants.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.chanjar.weixin.common.session; + +/** + * Manifest constants for the org.apache.catalina.session + * package. + * + * @author Craig R. McClanahan + */ + +public class Constants { + + public static final String Package = "me.chanjar.weixin.common.session"; + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSession.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSession.java new file mode 100644 index 00000000..79a7f4db --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSession.java @@ -0,0 +1,68 @@ +package me.chanjar.weixin.common.session; + +public interface InternalSession { + + /** + * Return the HttpSession for which this object + * is the facade. + */ + WxSession getSession(); + + /** + * Set the isValid flag for this session. + * + * @param isValid The new value for the isValid flag + */ + public void setValid(boolean isValid); + + /** + * Return the isValid flag for this session. + */ + boolean isValid(); + + /** + * Return the session identifier for this session. + */ + String getIdInternal(); + + /** + * Perform the internal processing required to invalidate this session, + * without triggering an exception if the session has already expired. + */ + void expire(); + + /** + * Update the accessed time information for this session. This method + * should be called by the context when a request comes in for a particular + * session, even if the application does not reference it. + */ + void access(); + + /** + * End the access. + */ + void endAccess(); + + /** + * Set the creation time for this session. This method is called by the + * Manager when an existing Session instance is reused. + * + * @param time The new creation time + */ + void setCreationTime(long time); + + /** + * Set the default maximum inactive interval (in seconds) + * for Sessions created by this Manager. + * + * @param interval The new default value + */ + void setMaxInactiveInterval(int interval); + + /** + * Set the session identifier for this session. + * + * @param id The new session identifier + */ + void setId(String id); +} 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 new file mode 100644 index 00000000..556d8ab8 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/InternalSessionManager.java @@ -0,0 +1,113 @@ +package me.chanjar.weixin.common.session; + +public interface InternalSessionManager { + + /** + * Return the active Session, associated with this Manager, with the + * specified session id (if any); otherwise return null. + * + * @param id The session id for the session to be returned + * + * @exception IllegalStateException if a new session cannot be + * instantiated for any reason + * @exception java.io.IOException if an input/output error occurs while + * processing this request + */ + InternalSession findSession(String id); + + /** + * Construct and return a new session object, based on the default + * settings specified by this Manager's properties. The session + * id specified will be used as the session id. + * If a new session cannot be created for any reason, return + * null. + * + * @param sessionId The session id which should be used to create the + * new session; if null, a new session id will be + * generated + * @exception IllegalStateException if a new session cannot be + * instantiated for any reason + */ + public InternalSession createSession(String sessionId); + + /** + * Remove this Session from the active Sessions for this Manager. + * + * @param session Session to be removed + */ + public void remove(InternalSession session); + + /** + * Remove this Session from the active Sessions for this Manager. + * + * @param session Session to be removed + * @param update Should the expiration statistics be updated + */ + public void remove(InternalSession session, boolean update); + + /** + * Add this Session to the set of active Sessions for this Manager. + * + * @param session Session to be added + */ + void add(InternalSession session); + + + /** + * Returns the number of active sessions + * + * @return number of sessions active + */ + int getActiveSessions(); + /** + * Get a session from the recycled ones or create a new empty one. + * The PersistentManager manager does not need to create session data + * because it reads it from the Store. + */ + InternalSession createEmptySession(); + + InternalSession[] findSessions(); + + /** + * Implements the Manager interface, direct call to processExpires + */ + 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); + + /** + *
+   * Set the manager checks frequency.
+   * 设置每尝试多少次清理过期session,才真的会执行一次清理动作
+   * 要和{@link #setBackgroundProcessorDelay(int)}联合起来看
+   * 如果把这个数字设置为6(默认),那么就是说manager要等待 6 * backgroundProcessorDay的时间才会清理过期session
+   * 
+ * @param processExpiresFrequency the new manager checks frequency + */ + void setProcessExpiresFrequency(int processExpiresFrequency); + + /** + *
+   * Set the manager background processor delay
+   * 设置manager sleep几秒,尝试执行一次background操作(清理过期session)
+   * 
+ * @param backgroundProcessorDelay + */ + void setBackgroundProcessorDelay(int backgroundProcessorDelay); + + + /** + * Set the maximum number of active Sessions allowed, or -1 for + * no limit. + * 设置最大活跃session数,默认无限 + * @param max The new maximum number of sessions + */ + void setMaxActiveSessions(int max); + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/LocalStrings.properties b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/LocalStrings.properties new file mode 100644 index 00000000..ede528f7 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/LocalStrings.properties @@ -0,0 +1,80 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +applicationSession.session.ise=invalid session state +applicationSession.value.iae=null value +fileStore.saving=Saving Session {0} to file {1} +fileStore.loading=Loading Session {0} from file {1} +fileStore.removing=Removing Session {0} at file {1} +fileStore.deleteFailed=Unable to delete file [{0}] which is preventing the creation of the session storage location +fileStore.createFailed=Unable to create directory [{0}] for the storage of session data +JDBCStore.close=Exception closing database connection {0} +JDBCStore.saving=Saving Session {0} to database {1} +JDBCStore.loading=Loading Session {0} from database {1} +JDBCStore.removing=Removing Session {0} at database {1} +JDBCStore.SQLException=SQL Error {0} +serverSession.value.iae=null value +sessionManagerImpl.createRandom=Created random number generator for session ID generation in {0}ms. +sessionManagerImpl.createSession.tmase=createSession: Too many active sessions +sessionManagerImpl.sessionTimeout=Invalid session timeout setting {0} +sessionManagerImpl.getSession.ise=getSession: Session id cannot be null +sessionManagerImpl.expireException=processsExpire: Exception during session expiration +sessionManagerImpl.loading=Loading persisted sessions from {0} +sessionManagerImpl.loading.cnfe=ClassNotFoundException while loading persisted sessions: {0} +sessionManagerImpl.loading.ioe=IOException while loading persisted sessions: {0} +sessionManagerImpl.unloading=Saving persisted sessions to {0} +sessionManagerImpl.unloading.debug=Unloading persisted sessions +sessionManagerImpl.unloading.ioe=IOException while saving persisted sessions: {0} +sessionManagerImpl.unloading.nosessions=No persisted sessions to unload +sessionManagerImpl.managerLoad=Exception loading sessions from persistent storage +sessionManagerImpl.managerUnload=Exception unloading sessions to persistent storage +sessionManagerImpl.createSession.ise=createSession: Session id cannot be null +sessionImpl.attributeEvent=Session attribute event listener threw exception +sessionImpl.bindingEvent=Session binding event listener threw exception +sessionImpl.invalidate.ise=invalidate: Session already invalidated +sessionImpl.isNew.ise=isNew: Session already invalidated +sessionImpl.getAttribute.ise=getAttribute: Session already invalidated +sessionImpl.getAttributeNames.ise=getAttributeNames: Session already invalidated +sessionImpl.getCreationTime.ise=getCreationTime: Session already invalidated +sessionImpl.getThisAccessedTime.ise=getThisAccessedTime: Session already invalidated +sessionImpl.getLastAccessedTime.ise=getLastAccessedTime: Session already invalidated +sessionImpl.getId.ise=getId: Session already invalidated +sessionImpl.getMaxInactiveInterval.ise=getMaxInactiveInterval: Session already invalidated +sessionImpl.getValueNames.ise=getValueNames: Session already invalidated +sessionImpl.logoutfail=Exception logging out user when expiring session +sessionImpl.notSerializable=Cannot serialize session attribute {0} for session {1} +sessionImpl.removeAttribute.ise=removeAttribute: Session already invalidated +sessionImpl.sessionEvent=Session event listener threw exception +sessionImpl.setAttribute.iae=setAttribute: Non-serializable attribute {0} +sessionImpl.setAttribute.ise=setAttribute: Session [{0}] has already been invalidated +sessionImpl.setAttribute.namenull=setAttribute: name parameter cannot be null +sessionImpl.sessionCreated=Created Session id = {0} +persistentManager.loading=Loading {0} persisted sessions +persistentManager.unloading=Saving {0} persisted sessions +persistentManager.expiring=Expiring {0} sessions before saving them +persistentManager.deserializeError=Error deserializing Session {0}: {1} +persistentManager.serializeError=Error serializing Session {0}: {1} +persistentManager.swapMaxIdle=Swapping session {0} to Store, idle for {1} seconds +persistentManager.backupMaxIdle=Backing up session {0} to Store, idle for {1} seconds +persistentManager.backupException=Exception occurred when backing up Session {0}: {1} +persistentManager.tooManyActive=Too many active sessions, {0}, looking for idle sessions to swap out +persistentManager.swapTooManyActive=Swapping out session {0}, idle for {1} seconds too many sessions active +persistentManager.processSwaps=Checking for sessions to swap out, {0} active sessions in memory +persistentManager.activeSession=Session {0} has been idle for {1} seconds +persistentManager.swapIn=Swapping session {0} in from Store +persistentManager.swapInException=Exception in the Store during swapIn: {0} +persistentManager.swapInInvalid=Swapped session {0} is invalid +persistentManager.storeKeysException=Unable to determine the list of session IDs for sessions in the session store, assuming that the store is empty +persistentManager.storeSizeException=Unable to determine the number of sessions in the session store, assuming that the store is empty diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSession.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSession.java new file mode 100644 index 00000000..d9827308 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSession.java @@ -0,0 +1,346 @@ +package me.chanjar.weixin.common.session; + +import me.chanjar.weixin.common.util.res.StringManager; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +public class StandardSession implements WxSession, InternalSession { + + /** + * The string manager for this package. + */ + protected static final StringManager sm = + StringManager.getManager(Constants.Package); + + // ------------------------------ WxSession + protected Map attributes = new ConcurrentHashMap(); + + @Override + public Object getAttribute(String name) { + + if (!isValidInternal()) + throw new IllegalStateException + (sm.getString("sessionImpl.getAttribute.ise")); + + if (name == null) return null; + + return (attributes.get(name)); + } + + @Override + public Enumeration getAttributeNames() { + if (!isValidInternal()) + throw new IllegalStateException + (sm.getString("sessionImpl.getAttributeNames.ise")); + + Set names = new HashSet(); + names.addAll(attributes.keySet()); + return Collections.enumeration(names); + } + + @Override + public void setAttribute(String name, Object value) { + // Name cannot be null + if (name == null) + throw new IllegalArgumentException + (sm.getString("sessionImpl.setAttribute.namenull")); + + // Null value is the same as removeAttribute() + if (value == null) { + removeAttribute(name); + return; + } + + // Validate our current state + if (!isValidInternal()) + throw new IllegalStateException(sm.getString( + "sessionImpl.setAttribute.ise", getIdInternal())); + + attributes.put(name, value); + + } + + + @Override + public void removeAttribute(String name) { + removeAttributeInternal(name); + } + + + @Override + public void invalidate() { + if (!isValidInternal()) + throw new IllegalStateException + (sm.getString("sessionImpl.invalidate.ise")); + + // Cause this session to expire + expire(); + + } + + // ------------------------------ InternalSession + /** + * The session identifier of this Session. + */ + protected String id = null; + + /** + * Flag indicating whether this session is valid or not. + */ + protected volatile boolean isValid = false; + + /** + * We are currently processing a session expiration, so bypass + * certain IllegalStateException tests. NOTE: This value is not + * included in the serialized version of this object. + */ + protected transient volatile boolean expiring = false; + + /** + * The Manager with which this Session is associated. + */ + protected transient InternalSessionManager manager = null; + + /** + * Type array. + */ + protected static final String EMPTY_ARRAY[] = new String[0]; + + /** + * The time this session was created, in milliseconds since midnight, + * January 1, 1970 GMT. + */ + protected long creationTime = 0L; + + /** + * The current accessed time for this session. + */ + protected volatile long thisAccessedTime = creationTime; + + /** + * The default maximum inactive interval for Sessions created by + * this Manager. + */ + protected int maxInactiveInterval = 30 * 60; + + /** + * The facade associated with this session. NOTE: This value is not + * included in the serialized version of this object. + */ + protected transient StandardSessionFacade facade = null; + + /** + * The access count for this session. + */ + protected transient AtomicInteger accessCount = null; + + + public StandardSession(InternalSessionManager manager) { + this.manager = manager; + this.accessCount = new AtomicInteger(); + } + + + @Override + public WxSession getSession() { + + if (facade == null){ + facade = new StandardSessionFacade(this); + } + return (facade); + + } + + /** + * Return the isValid flag for this session without any expiration + * check. + */ + protected boolean isValidInternal() { + return this.isValid; + } + + /** + * Set the isValid flag for this session. + * + * @param isValid The new value for the isValid flag + */ + @Override + public void setValid(boolean isValid) { + this.isValid = isValid; + } + + @Override + public boolean isValid() { + if (!this.isValid) { + return false; + } + + if (this.expiring) { + return true; + } + + if (accessCount.get() > 0) { + return true; + } + + if (maxInactiveInterval > 0) { + long timeNow = System.currentTimeMillis(); + int timeIdle; + timeIdle = (int) ((timeNow - thisAccessedTime) / 1000L); + if (timeIdle >= maxInactiveInterval) { + expire(); + } + } + + return this.isValid; + } + + @Override + public String getIdInternal() { + return (this.id); + } + + protected void removeAttributeInternal(String name) { + // Avoid NPE + if (name == null) return; + + // Remove this attribute from our collection + attributes.remove(name); + + } + + @Override + public void expire() { + // Check to see if session has already been invalidated. + // Do not check expiring at this point as expire should not return until + // isValid is false + if (!isValid) + return; + + synchronized (this) { + // Check again, now we are inside the sync so this code only runs once + // Double check locking - isValid needs to be volatile + // The check of expiring is to ensure that an infinite loop is not + // entered as per bug 56339 + if (expiring || !isValid) + return; + + if (manager == null) + return; + + // Mark this session as "being expired" + expiring = true; + + accessCount.set(0); + + // Remove this session from our manager's active sessions + manager.remove(this, true); + + + // We have completed expire of this session + setValid(false); + expiring = false; + + // Unbind any objects associated with this session + String keys[] = keys(); + for (int i = 0; i < keys.length; i++) { + removeAttributeInternal(keys[i]); + } + } + + + } + + + @Override + public void access() { + + this.thisAccessedTime = System.currentTimeMillis(); + accessCount.incrementAndGet(); + + } + + + @Override + public void endAccess() { + + this.thisAccessedTime = System.currentTimeMillis(); + accessCount.decrementAndGet(); + + } + + @Override + public void setCreationTime(long time) { + + this.creationTime = time; + this.thisAccessedTime = time; + + } + + @Override + public void setMaxInactiveInterval(int interval) { + this.maxInactiveInterval = interval; + } + + + @Override + public void setId(String id) { + if ((this.id != null) && (manager != null)) + manager.remove(this); + + this.id = id; + + if (manager != null) + manager.add(this); + } + + /** + * Return the names of all currently defined session attributes + * as an array of Strings. If there are no defined attributes, a + * zero-length array is returned. + */ + protected String[] keys() { + + return attributes.keySet().toArray(EMPTY_ARRAY); + + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof StandardSession)) return false; + + StandardSession session = (StandardSession) 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/main/java/me/chanjar/weixin/common/session/StandardSessionFacade.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSessionFacade.java new file mode 100644 index 00000000..a9631de6 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSessionFacade.java @@ -0,0 +1,45 @@ +package me.chanjar.weixin.common.session; + +import java.util.Enumeration; + +public class StandardSessionFacade implements WxSession { + + /** + * Wrapped session object. + */ + private WxSession session = null; + + public StandardSessionFacade(StandardSession session) { + this.session = session; + } + + public InternalSession getInternalSession() { + return (InternalSession) session; + } + + @Override + public Object getAttribute(String name) { + return session.getAttribute(name); + } + + @Override + public Enumeration getAttributeNames() { + return session.getAttributeNames(); + } + + @Override + public void setAttribute(String name, Object value) { + session.setAttribute(name, value); + } + + @Override + public void removeAttribute(String name) { + session.removeAttribute(name); + } + + @Override + public void invalidate() { + session.invalidate(); + } + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSessionManager.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSessionManager.java new file mode 100644 index 00000000..f3a02695 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/StandardSessionManager.java @@ -0,0 +1,321 @@ +package me.chanjar.weixin.common.session; + +import me.chanjar.weixin.common.util.res.StringManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * 基于内存的session manager + */ +public class StandardSessionManager implements WxSessionManager, InternalSessionManager { + + protected final Logger log = LoggerFactory.getLogger(StandardSessionManager.class); + + protected static final StringManager sm = + StringManager.getManager(Constants.Package); + + /** + * The set of currently active Sessions for this Manager, keyed by + * session identifier. + */ + protected Map sessions = new ConcurrentHashMap(); + + @Override + public WxSession getSession(String sessionId) { + return getSession(sessionId, true); + } + + @Override + public WxSession getSession(String sessionId, boolean create) { + if (sessionId == null) { + throw new IllegalStateException + (sm.getString("sessionManagerImpl.getSession.ise")); + } + + InternalSession session = findSession(sessionId); + if ((session != null) && !session.isValid()) { + session = null; + } + if (session != null) { + session.access(); + return session.getSession(); + } + + // Create a new session if requested and the response is not committed + if (!create) { + return (null); + } + + session = createSession(sessionId); + + if (session == null) { + return null; + } + + session.access(); + return session.getSession(); + } + + + // -------------------------------------- InternalSessionManager + /** + * The descriptive name of this Manager implementation (for logging). + */ + private static final String name = "SessionManagerImpl"; + + /** + * The maximum number of active Sessions allowed, or -1 for no limit. + */ + protected int maxActiveSessions = -1; + + /** + * Number of session creations that failed due to maxActiveSessions. + */ + protected int rejectedSessions = 0; + + /** + * The default maximum inactive interval for Sessions created by + * this Manager. + */ + protected int maxInactiveInterval = 30 * 60; + + // Number of sessions created by this manager + protected long sessionCounter=0; + + protected volatile int maxActive=0; + + private final Object maxActiveUpdateLock = new Object(); + + /** + * Processing time during session expiration. + */ + protected long processingTime = 0; + + /** + * Iteration count for background processing. + */ + private int count = 0; + + /** + * Frequency of the session expiration, and related manager operations. + * Manager operations will be done once for the specified amount of + * backgrondProcess calls (ie, the lower the amount, the most often the + * checks will occur). + */ + protected int processExpiresFrequency = 6; + + /** + * background processor delay in seconds + */ + protected int backgroundProcessorDelay = 10; + + /** + * 后台清理线程是否已经开启 + */ + private final AtomicBoolean backgroundProcessStarted = new AtomicBoolean(false); + + @Override + public void remove(InternalSession session) { + remove(session, false); + } + + @Override + public void remove(InternalSession session, boolean update) { + if (session.getIdInternal() != null) { + sessions.remove(session.getIdInternal()); + } + } + + + + @Override + public InternalSession findSession(String id) { + + if (id == null) + return (null); + return sessions.get(id); + + } + + @Override + public InternalSession createSession(String sessionId) { + if (sessionId == null) { + throw new IllegalStateException + (sm.getString("sessionManagerImpl.createSession.ise")); + } + + if ((maxActiveSessions >= 0) && + (getActiveSessions() >= maxActiveSessions)) { + rejectedSessions++; + throw new TooManyActiveSessionsException( + sm.getString("sessionManagerImpl.createSession.tmase"), + maxActiveSessions); + } + + // Recycle or create a Session instance + InternalSession session = createEmptySession(); + + // Initialize the properties of the new session and return it + session.setValid(true); + session.setCreationTime(System.currentTimeMillis()); + session.setMaxInactiveInterval(this.maxInactiveInterval); + String id = sessionId; + session.setId(id); + sessionCounter++; + + return (session); + + } + + + @Override + public int getActiveSessions() { + return sessions.size(); + } + + + @Override + public InternalSession createEmptySession() { + return (getNewSession()); + } + + /** + * Get new session class to be used in the doLoad() method. + */ + protected InternalSession getNewSession() { + return new StandardSession(this); + } + + + @Override + public void add(InternalSession session) { + + // 当第一次有session创建的时候,开启session清理线程 + if (!backgroundProcessStarted.getAndSet(true)) { + Thread t = new Thread(new Runnable() { + @Override + public void run() { + while (true) { + try { + // 每秒清理一次 + Thread.sleep(backgroundProcessorDelay * 1000l); + backgroundProcess(); + } catch (InterruptedException e) { + log.error("SessionManagerImpl.backgroundProcess error", e); + } + } + } + }); + t.setDaemon(true); + t.start(); + } + + sessions.put(session.getIdInternal(), session); + int size = getActiveSessions(); + if( size > maxActive ) { + synchronized(maxActiveUpdateLock) { + if( size > maxActive ) { + maxActive = size; + } + } + } + + } + + /** + * Return the set of active Sessions associated with this Manager. + * If this Manager has no active Sessions, a zero-length array is returned. + */ + @Override + public InternalSession[] findSessions() { + + return sessions.values().toArray(new InternalSession[0]); + + } + + @Override + public void backgroundProcess() { + count = (count + 1) % processExpiresFrequency; + if (count == 0) + processExpires(); + } + + /** + * Invalidate all sessions that have expired. + */ + public void processExpires() { + + long timeNow = System.currentTimeMillis(); + InternalSession sessions[] = findSessions(); + int expireHere = 0 ; + + if(log.isDebugEnabled()) + log.debug("Start expire sessions {} at {} sessioncount {}", getName(), timeNow, sessions.length); + for (int i = 0; i < sessions.length; i++) { + if (sessions[i]!=null && !sessions[i].isValid()) { + expireHere++; + } + } + long timeEnd = System.currentTimeMillis(); + if(log.isDebugEnabled()) + log.debug("End expire sessions {} processingTime {} expired sessions: {}", getName(), timeEnd - timeNow, expireHere); + processingTime += ( timeEnd - timeNow ); + + } + + + @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. + */ + public String getName() { + + return (name); + + } + + /** + * Set the maximum number of active Sessions allowed, or -1 for + * no limit. + * + * @param max The new maximum number of sessions + */ + @Override + public void setMaxActiveSessions(int max) { + + this.maxActiveSessions = max; + + } + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/TooManyActiveSessionsException.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/TooManyActiveSessionsException.java new file mode 100644 index 00000000..bd0c0f2b --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/TooManyActiveSessionsException.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package me.chanjar.weixin.common.session; + +/** + * An exception that indicates the maximum number of active sessions has been + * reached and the server is refusing to create any new sessions. + */ +public class TooManyActiveSessionsException + extends IllegalStateException +{ + private static final long serialVersionUID = 1L; + + /** + * The maximum number of active sessions the server will tolerate. + */ + private final int maxActiveSessions; + + /** + * Creates a new TooManyActiveSessionsException. + * + * @param message A description for the exception. + * @param maxActive The maximum number of active sessions allowed by the + * session manager. + */ + public TooManyActiveSessionsException(String message, + int maxActive) + { + super(message); + + maxActiveSessions = maxActive; + } + + /** + * Gets the maximum number of sessions allowed by the session manager. + * + * @return The maximum number of sessions allowed by the session manager. + */ + public int getMaxActiveSessions() + { + return maxActiveSessions; + } +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/WxSession.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/WxSession.java new file mode 100644 index 00000000..816b172b --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/WxSession.java @@ -0,0 +1,17 @@ +package me.chanjar.weixin.common.session; + +import java.util.Enumeration; + +public interface WxSession { + + public Object getAttribute(String name); + + public Enumeration getAttributeNames(); + + public void setAttribute(String name, Object value); + + public void removeAttribute(String name); + + public void invalidate(); + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/WxSessionManager.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/WxSessionManager.java new file mode 100644 index 00000000..262d9fe4 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/session/WxSessionManager.java @@ -0,0 +1,21 @@ +package me.chanjar.weixin.common.session; + +public interface WxSessionManager { + + /** + * 获取某个sessionId对应的session,如果sessionId没有对应的session,则新建一个并返回。 + * @param sessionId + * @return + */ + public WxSession getSession(String sessionId); + + /** + * 获取某个sessionId对应的session,如果sessionId没有对应的session,若create为true则新建一个,否则返回null。 + * @param sessionId + * @param create + * @return + */ + public WxSession getSession(String sessionId, boolean create); + + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMessageDuplicateChecker.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMessageDuplicateChecker.java new file mode 100644 index 00000000..1abe588b --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMessageDuplicateChecker.java @@ -0,0 +1,27 @@ +package me.chanjar.weixin.common.util; + +/** + *
+ * 消息重复检查器
+ * 微信服务器在五秒内收不到响应会断掉连接,并且重新发起请求,总共重试三次
+ * 
+ */ +public interface WxMessageDuplicateChecker { + + /** + *

公众号的排重方式

+ * + *

普通消息:关于重试的消息排重,推荐使用msgid排重。文档参考

+ *

事件消息:关于重试的消息排重,推荐使用FromUserName + CreateTime 排重。文档参考

+ * + *

企业号的排重方式

+ * + * 官方文档完全没有写,参照公众号的方式排重。 + * + *

或者可以采取更简单的方式,如果有MsgId就用MsgId排重,如果没有就用FromUserName+CreateTime排重

+ * @param messageId messageId需要根据上面讲的方式构造 + * @return 如果是重复消息,返回true,否则返回false + */ + public boolean isDuplicate(String messageId); + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMessageInMemoryDuplicateChecker.java similarity index 63% rename from weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java rename to weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMessageInMemoryDuplicateChecker.java index 0c8eee91..ae0e0623 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMessageInMemoryDuplicateChecker.java @@ -2,6 +2,7 @@ package me.chanjar.weixin.common.util; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; /** *
@@ -9,7 +10,7 @@ import java.util.concurrent.ConcurrentHashMap;
  * 将每个消息id保存在内存里,每隔5秒清理已经过期的消息id,每个消息id的过期时间是15秒
  * 
*/ -public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker { +public class WxMessageInMemoryDuplicateChecker implements WxMessageDuplicateChecker { /** * 一个消息ID在内存的过期时间:15秒 @@ -21,7 +22,15 @@ public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker */ private final Long clearPeriod; - private final ConcurrentHashMap msgId2Timestamp = new ConcurrentHashMap(); + /** + * 消息id->消息时间戳的map + */ + private final ConcurrentHashMap msgId2Timestamp = new ConcurrentHashMap(); + + /** + * 后台清理线程是否已经开启 + */ + private final AtomicBoolean backgroundProcessStarted = new AtomicBoolean(false); /** * WxMsgIdInMemoryDuplicateChecker构造函数 @@ -30,10 +39,9 @@ public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker * 每隔多少周期检查消息ID是否过期:5秒 * */ - public WxMsgIdInMemoryDuplicateChecker() { + public WxMessageInMemoryDuplicateChecker() { this.timeToLive = 15 * 1000l; this.clearPeriod = 5 * 1000l; - this.start(); } /** @@ -41,13 +49,15 @@ public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker * @param timeToLive 一个消息ID在内存的过期时间:毫秒 * @param clearPeriod 每隔多少周期检查消息ID是否过期:毫秒 */ - public WxMsgIdInMemoryDuplicateChecker(Long timeToLive, Long clearPeriod) { + public WxMessageInMemoryDuplicateChecker(Long timeToLive, Long clearPeriod) { this.timeToLive = timeToLive; this.clearPeriod = clearPeriod; - this.start(); } - private void start() { + protected void checkBackgroundProcessStarted() { + if (backgroundProcessStarted.getAndSet(true)) { + return; + } Thread t = new Thread(new Runnable() { @Override public void run() { @@ -55,7 +65,7 @@ public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker while (true) { Thread.sleep(clearPeriod); Long now = System.currentTimeMillis(); - for (Map.Entry entry : msgId2Timestamp.entrySet()) { + for (Map.Entry entry : msgId2Timestamp.entrySet()) { if (now - entry.getValue() > timeToLive) { msgId2Timestamp.entrySet().remove(entry); } @@ -71,8 +81,12 @@ public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker } @Override - public boolean isDuplicate(Long wxMsgId) { - Long timestamp = msgId2Timestamp.putIfAbsent(wxMsgId, System.currentTimeMillis()); + public boolean isDuplicate(String messageId) { + if (messageId == null) { + return false; + } + checkBackgroundProcessStarted(); + Long timestamp = msgId2Timestamp.putIfAbsent(messageId, System.currentTimeMillis()); if (timestamp == null) { // 第一次接收到这个消息 return false; diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java deleted file mode 100644 index b3d320b1..00000000 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java +++ /dev/null @@ -1,18 +0,0 @@ -package me.chanjar.weixin.common.util; - -/** - *
- * 消息重复检查器
- * 微信服务器在五秒内收不到响应会断掉连接,并且重新发起请求,总共重试三次
- * 
- */ -public interface WxMsgIdDuplicateChecker { - - /** - * 检查消息ID是否重复 - * @param wxMsgId - * @return 如果是重复消息,返回true,否则返回false - */ - public boolean isDuplicate(Long wxMsgId); - -} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/res/StringManager.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/res/StringManager.java new file mode 100644 index 00000000..4668ec25 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/res/StringManager.java @@ -0,0 +1,251 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package me.chanjar.weixin.common.util.res; + +import java.text.MessageFormat; +import java.util.Enumeration; +import java.util.Hashtable; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.MissingResourceException; +import java.util.ResourceBundle; + +/** + * An internationalization / localization helper class which reduces + * the bother of handling ResourceBundles and takes care of the + * common cases of message formating which otherwise require the + * creation of Object arrays and such. + * + *

The StringManager operates on a package basis. One StringManager + * per package can be created and accessed via the getManager method + * call. + * + *

The StringManager will look for a ResourceBundle named by + * the package name given plus the suffix of "LocalStrings". In + * practice, this means that the localized information will be contained + * in a LocalStrings.properties file located in the package + * directory of the classpath. + * + *

Please see the documentation for java.util.ResourceBundle for + * more information. + * + * @author James Duncan Davidson [duncan@eng.sun.com] + * @author James Todd [gonzo@eng.sun.com] + * @author Mel Martinez [mmartinez@g1440.com] + * @see java.util.ResourceBundle + */ +public class StringManager { + + private static int LOCALE_CACHE_SIZE = 10; + + /** + * The ResourceBundle for this StringManager. + */ + private final ResourceBundle bundle; + private final Locale locale; + + /** + * Creates a new StringManager for a given package. This is a + * private method and all access to it is arbitrated by the + * static getManager method call so that only one StringManager + * per package will be created. + * + * @param packageName Name of package to create StringManager for. + */ + private StringManager(String packageName, Locale locale) { + String bundleName = packageName + ".LocalStrings"; + ResourceBundle bnd = null; + try { + bnd = ResourceBundle.getBundle(bundleName, locale); + } catch( MissingResourceException ex ) { + // Try from the current loader (that's the case for trusted apps) + // Should only be required if using a TC5 style classloader structure + // where common != shared != server + ClassLoader cl = Thread.currentThread().getContextClassLoader(); + if( cl != null ) { + try { + bnd = ResourceBundle.getBundle(bundleName, locale, cl); + } catch(MissingResourceException ex2) { + // Ignore + } + } + } + bundle = bnd; + // Get the actual locale, which may be different from the requested one + if (bundle != null) { + Locale bundleLocale = bundle.getLocale(); + if (bundleLocale.equals(Locale.ROOT)) { + this.locale = Locale.ENGLISH; + } else { + this.locale = bundleLocale; + } + } else { + this.locale = null; + } + } + + /** + Get a string from the underlying resource bundle or return + null if the String is not found. + + @param key to desired resource String + @return resource String matching key from underlying + bundle or null if not found. + @throws IllegalArgumentException if key is null. + */ + public String getString(String key) { + if(key == null){ + String msg = "key may not have a null value"; + + throw new IllegalArgumentException(msg); + } + + String str = null; + + try { + // Avoid NPE if bundle is null and treat it like an MRE + if (bundle != null) { + str = bundle.getString(key); + } + } catch(MissingResourceException mre) { + //bad: shouldn't mask an exception the following way: + // str = "[cannot find message associated with key '" + key + + // "' due to " + mre + "]"; + // because it hides the fact that the String was missing + // from the calling code. + //good: could just throw the exception (or wrap it in another) + // but that would probably cause much havoc on existing + // code. + //better: consistent with container pattern to + // simply return null. Calling code can then do + // a null check. + str = null; + } + + return str; + } + + /** + * Get a string from the underlying resource bundle and format + * it with the given set of arguments. + * + * @param key + * @param args + */ + public String getString(final String key, final Object... args) { + String value = getString(key); + if (value == null) { + value = key; + } + + MessageFormat mf = new MessageFormat(value); + mf.setLocale(locale); + return mf.format(args, new StringBuffer(), null).toString(); + } + + /** + * Identify the Locale this StringManager is associated with + */ + public Locale getLocale() { + return locale; + } + + // -------------------------------------------------------------- + // STATIC SUPPORT METHODS + // -------------------------------------------------------------- + + private static final Map> managers = + new Hashtable>(); + + /** + * Get the StringManager for a particular package. If a manager for + * a package already exists, it will be reused, else a new + * StringManager will be created and returned. + * + * @param packageName The package name + */ + public static final synchronized StringManager getManager( + String packageName) { + return getManager(packageName, Locale.getDefault()); + } + + /** + * Get the StringManager for a particular package and Locale. If a manager + * for a package/Locale combination already exists, it will be reused, else + * a new StringManager will be created and returned. + * + * @param packageName The package name + * @param locale The Locale + */ + public static final synchronized StringManager getManager( + String packageName, Locale locale) { + + Map map = managers.get(packageName); + if (map == null) { + /* + * Don't want the HashMap to be expanded beyond LOCALE_CACHE_SIZE. + * Expansion occurs when size() exceeds capacity. Therefore keep + * size at or below capacity. + * removeEldestEntry() executes after insertion therefore the test + * for removal needs to use one less than the maximum desired size + * + */ + map = new LinkedHashMap(LOCALE_CACHE_SIZE, 1, true) { + private static final long serialVersionUID = 1L; + @Override + protected boolean removeEldestEntry( + Map.Entry eldest) { + if (size() > (LOCALE_CACHE_SIZE - 1)) { + return true; + } + return false; + } + }; + managers.put(packageName, map); + } + + StringManager mgr = map.get(locale); + if (mgr == null) { + mgr = new StringManager(packageName, locale); + map.put(locale, mgr); + } + return mgr; + } + + /** + * Retrieve the StringManager for a list of Locales. The first StringManager + * found will be returned. + * + * @param requestedLocales the list of Locales + * + * @return the found StringManager or the default StringManager + */ + public static StringManager getManager(String packageName, + Enumeration requestedLocales) { + while (requestedLocales.hasMoreElements()) { + Locale locale = requestedLocales.nextElement(); + StringManager result = getManager(packageName, locale); + if (result.getLocale().equals(locale)) { + return result; + } + } + // Return the default + return getManager(packageName); + } +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamCDataConverter.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamCDataConverter.java index 69bad096..aa948b34 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamCDataConverter.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamCDataConverter.java @@ -2,9 +2,6 @@ package me.chanjar.weixin.common.util.xml; import com.thoughtworks.xstream.converters.basic.StringConverter; -/** - * Created by qianjia on 15/1/19. - */ public class XStreamCDataConverter extends StringConverter { @Override diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamInitializer.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamInitializer.java index 163c6558..d545a796 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamInitializer.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamInitializer.java @@ -14,9 +14,6 @@ import com.thoughtworks.xstream.security.PrimitiveTypePermission; import java.io.Writer; -/** - * Created by qianjia on 15/1/19. - */ public class XStreamInitializer { public static XStream getInstance() { diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamMediaIdConverter.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamMediaIdConverter.java index 0e375ea5..f67a3d9d 100644 --- a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamMediaIdConverter.java +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/xml/XStreamMediaIdConverter.java @@ -1,8 +1,5 @@ package me.chanjar.weixin.common.util.xml; -/** - * Created by qianjia on 15/1/19. - */ public class XStreamMediaIdConverter extends XStreamCDataConverter { @Override public String toString(Object obj) { diff --git a/weixin-java-common/src/test/java/me/chanjar/weixin/common/session/SessionTest.java b/weixin-java-common/src/test/java/me/chanjar/weixin/common/session/SessionTest.java new file mode 100644 index 00000000..bcb029b1 --- /dev/null +++ b/weixin-java-common/src/test/java/me/chanjar/weixin/common/session/SessionTest.java @@ -0,0 +1,132 @@ +package me.chanjar.weixin.common.session; + +import org.testng.Assert; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Test +public class SessionTest { + + @DataProvider + public Object[][] getSessionManager() { + + return new Object[][] { + new Object[] { new StandardSessionManager() } + }; + + } + + + @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.assertEquals(session1, session2); + Assert.assertTrue(session1 == session2); + + WxSession abc1 = sessionManager.getSession("abc1"); + Assert.assertNotEquals(session1, abc1); + + WxSession abc1b = sessionManager.getSession("abc1", false); + Assert.assertEquals(abc1, abc1b); + + WxSession def = sessionManager.getSession("def", false); + Assert.assertNull(def); + + } + + @Test(dataProvider = "getSessionManager") + public void testInvalidateAngGet(WxSessionManager sessionManager) { + + WxSession session1 = sessionManager.getSession("abc"); + session1.invalidate(); + WxSession session2 = sessionManager.getSession("abc"); + Assert.assertNotEquals(session1, session2); + InternalSessionManager ism = (InternalSessionManager) sessionManager; + Assert.assertEquals(ism.getActiveSessions(), 1); + + } + + @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); + + } + + @Test(dataProvider = "getSessionManager") + public void testBackgroundProcess2(WxSessionManager sessionManager) throws InterruptedException { + + InternalSessionManager ism = (InternalSessionManager) sessionManager; + ism.setMaxInactiveInterval(100); + ism.setProcessExpiresFrequency(1); + ism.setBackgroundProcessorDelay(1); + + Assert.assertEquals(ism.getActiveSessions(), 0); + + InternalSession abc = ism.createSession("abc"); + abc.setMaxInactiveInterval(1); + abc.endAccess(); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + + } + + @Test(dataProvider = "getSessionManager") + public void testMaxActive(WxSessionManager sessionManager) throws InterruptedException { + + InternalSessionManager ism = (InternalSessionManager) sessionManager; + ism.setMaxActiveSessions(2); + + ism.createSession("abc"); + ism.createSession("abc"); + ism.createSession("def"); + + } + + @Test(dataProvider = "getSessionManager", expectedExceptions = TooManyActiveSessionsException.class) + public void testMaxActive2(WxSessionManager sessionManager) throws InterruptedException { + + InternalSessionManager ism = (InternalSessionManager) sessionManager; + ism.setMaxActiveSessions(2); + + ism.createSession("abc"); + ism.createSession("abc"); + ism.createSession("def"); + ism.createSession("xyz"); + + } +} diff --git a/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java b/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMessageInMemoryDuplicateCheckerTest.java similarity index 61% rename from weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java rename to weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMessageInMemoryDuplicateCheckerTest.java index a35eba5c..28642c44 100644 --- a/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java +++ b/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMessageInMemoryDuplicateCheckerTest.java @@ -3,33 +3,30 @@ package me.chanjar.weixin.common.util; import org.testng.Assert; import org.testng.annotations.Test; -/** - * Created by qianjia on 15/1/20. - */ @Test -public class WxMsgIdInMemoryDuplicateCheckerTest { +public class WxMessageInMemoryDuplicateCheckerTest { public void test() throws InterruptedException { Long[] msgIds = new Long[] { 1l, 2l, 3l, 4l, 5l, 6l, 7l, 8l }; - WxMsgIdInMemoryDuplicateChecker checker = new WxMsgIdInMemoryDuplicateChecker(2000l, 1000l); + WxMessageInMemoryDuplicateChecker checker = new WxMessageInMemoryDuplicateChecker(2000l, 1000l); // 第一次检查 for (Long msgId : msgIds) { - boolean result = checker.isDuplicate(msgId); + boolean result = checker.isDuplicate(String.valueOf(msgId)); Assert.assertFalse(result); } // 过1秒再检查 Thread.sleep(1000l); for (Long msgId : msgIds) { - boolean result = checker.isDuplicate(msgId); + boolean result = checker.isDuplicate(String.valueOf(msgId)); Assert.assertTrue(result); } // 过1.5秒再检查 Thread.sleep(1500l); for (Long msgId : msgIds) { - boolean result = checker.isDuplicate(msgId); + boolean result = checker.isDuplicate(String.valueOf(msgId)); Assert.assertFalse(result); } diff --git a/weixin-java-common/src/test/resources/logback-test.xml b/weixin-java-common/src/test/resources/logback-test.xml new file mode 100644 index 00000000..c0d6fc3c --- /dev/null +++ b/weixin-java-common/src/test/resources/logback-test.xml @@ -0,0 +1,16 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + diff --git a/weixin-java-common/src/test/resources/testng.xml b/weixin-java-common/src/test/resources/testng.xml index 49d65248..f2222275 100644 --- a/weixin-java-common/src/test/resources/testng.xml +++ b/weixin-java-common/src/test/resources/testng.xml @@ -1,13 +1,14 @@ - - - - - - - - - + + + + + + + + + + diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpInMemoryConfigStorage.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpInMemoryConfigStorage.java index dd4882e2..0a35fcd9 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpInMemoryConfigStorage.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpInMemoryConfigStorage.java @@ -37,11 +37,11 @@ public class WxCpInMemoryConfigStorage implements WxCpConfigStorage { this.expiresTime = 0; } - public void updateAccessToken(WxAccessToken accessToken) { + public synchronized void updateAccessToken(WxAccessToken accessToken) { updateAccessToken(accessToken.getAccessToken(), accessToken.getExpiresIn()); } - public void updateAccessToken(String accessToken, int expiresInSeconds) { + public synchronized void updateAccessToken(String accessToken, int expiresInSeconds) { this.accessToken = accessToken; this.expiresTime = System.currentTimeMillis() + (expiresInSeconds - 200) * 1000l; } 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 e4e26eb2..f9a3c289 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,7 @@ package me.chanjar.weixin.cp.api; +import me.chanjar.weixin.common.session.WxSession; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import me.chanjar.weixin.cp.bean.WxCpXmlOutMessage; @@ -7,18 +9,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 5cb478e3..92fa41f1 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 c88bc8a3..34b36ec3 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,21 @@ package me.chanjar.weixin.cp.api; -import me.chanjar.weixin.common.util.WxMsgIdDuplicateChecker; -import me.chanjar.weixin.common.util.WxMsgIdInMemoryDuplicateChecker; +import me.chanjar.weixin.common.session.*; +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,7 +48,9 @@ import java.util.regex.Pattern; */ public class WxCpMessageRouter { - private static final int DEFAULT_THREAD_POOL_SIZE = 20; + protected final Logger log = LoggerFactory.getLogger(WxCpMessageRouter.class); + + private static final int DEFAULT_THREAD_POOL_SIZE = 100; private final List rules = new ArrayList(); @@ -51,22 +58,22 @@ public class WxCpMessageRouter { private ExecutorService executorService; - private WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker; + private WxMessageDuplicateChecker messageDuplicateChecker; + + private WxSessionManager sessionManager; public WxCpMessageRouter(WxCpService wxCpService) { this.wxCpService = wxCpService; this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE); - this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); - } - - public WxCpMessageRouter(WxCpService wxMpService, int threadPoolSize) { - this.wxCpService = wxMpService; - this.executorService = Executors.newFixedThreadPool(threadPoolSize); - this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); + this.messageDuplicateChecker = new WxMessageInMemoryDuplicateChecker(); + this.sessionManager = new StandardSessionManager(); } /** - * 设置自定义的ExecutorService + *

+   * 设置自定义的 {@link ExecutorService}
+   * 如果不调用该方法,默认使用 Executors.newFixedThreadPool(100)
+   * 
* @param executorService */ public void setExecutorService(ExecutorService executorService) { @@ -74,11 +81,25 @@ public class WxCpMessageRouter { } /** - * 设置自定义的WxMsgIdDuplicateChecker - * @param wxMsgIdDuplicateChecker + *
+   * 设置自定义的 {@link me.chanjar.weixin.common.util.WxMessageDuplicateChecker}
+   * 如果不调用该方法,默认使用 {@link me.chanjar.weixin.common.util.WxMessageInMemoryDuplicateChecker}
+   * 
+ * @param messageDuplicateChecker + */ + public void setMessageDuplicateChecker(WxMessageDuplicateChecker messageDuplicateChecker) { + this.messageDuplicateChecker = messageDuplicateChecker; + } + + /** + *
+   * 设置自定义的{@link me.chanjar.weixin.common.session.WxSessionManager}
+   * 如果不调用该方法,默认使用 {@linke SessionManagerImpl}
+   * 
+ * @param sessionManager */ - public void setWxMsgIdDuplicateChecker(WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker) { - this.wxMsgIdDuplicateChecker = wxMsgIdDuplicateChecker; + public void setSessionManager(WxSessionManager sessionManager) { + this.sessionManager = sessionManager; } /** @@ -86,7 +107,7 @@ public class WxCpMessageRouter { * @return */ public Rule rule() { - return new Rule(this, wxCpService); + return new Rule(this, wxCpService, sessionManager); } /** @@ -94,7 +115,7 @@ public class WxCpMessageRouter { * @param wxMessage */ public WxCpXmlOutMessage route(final WxCpXmlMessage wxMessage) { - if (wxMsgIdDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { + if (isDuplicateMessage(wxMessage)) { // 如果是重复消息,那么就不做处理 return null; } @@ -115,27 +136,84 @@ 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访问结束 + log.debug("End session access: async=false, sessionId={}", wxMessage.getFromUserName()); + sessionEndAccess(wxMessage); } } + + if (futures.size() > 0) { + executorService.submit(new Runnable() { + @Override + public void run() { + for (Future future : futures) { + try { + future.get(); + log.debug("End session access: async=true, sessionId={}", wxMessage.getFromUserName()); + // 异步操作结束,session访问结束 + sessionEndAccess(wxMessage); + } catch (InterruptedException e) { + log.error("Error happened when wait task finish", e); + } catch (ExecutionException e) { + log.error("Error happened when wait task finish", e); + } + } + } + }); + } return res; } + protected boolean isDuplicateMessage(WxCpXmlMessage wxMessage) { + + String messageId = ""; + if (wxMessage.getMsgId() == null) { + messageId = wxMessage.getFromUserName() + "-" + String.valueOf(wxMessage.getCreateTime()); + } else { + messageId = String.valueOf(wxMessage.getMsgId()); + } + + if (messageDuplicateChecker.isDuplicate(messageId)) { + return true; + } + return false; + + } + + /** + * 对session的访问结束 + * @param wxMessage + */ + protected void sessionEndAccess(WxCpXmlMessage wxMessage) { + + InternalSession session = ((InternalSessionManager)sessionManager).findSession(wxMessage.getFromUserName()); + if (session != null) { + 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; @@ -158,9 +236,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; } /** @@ -338,7 +417,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; } } @@ -347,7 +426,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 2c25614a..21669b19 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 @@ -3,6 +3,8 @@ package me.chanjar.weixin.cp.api; import me.chanjar.weixin.common.bean.WxMenu; import me.chanjar.weixin.common.bean.result.WxMediaUploadResult; import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.session.WxSession; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.http.RequestExecutor; import me.chanjar.weixin.cp.bean.WxCpDepart; import me.chanjar.weixin.cp.bean.WxCpMessage; @@ -353,4 +355,46 @@ public interface WxCpService { * @param wxConfigProvider */ public void setWxCpConfigStorage(WxCpConfigStorage wxConfigProvider); + + /** + *
+   * 设置当微信系统响应系统繁忙时,要等待多少 retrySleepMillis(ms) * 2^(重试次数 - 1) 再发起重试
+   * 默认:1000ms
+   * 
+ * @param retrySleepMillis + */ + void setRetrySleepMillis(int retrySleepMillis); + + /** + *
+   * 设置当微信系统响应系统繁忙时,最大重试次数
+   * 默认:5次
+   * 
+ * @param maxRetryTimes + */ + void setMaxRetryTimes(int maxRetryTimes); + + /** + * 获取某个sessionId对应的session,如果sessionId没有对应的session,则新建一个并返回。 + * @param id id可以为任意字符串,建议使用FromUserName作为id + * @return + */ + WxSession getSession(String id); + + /** + * 获取某个sessionId对应的session,如果sessionId没有对应的session,若create为true则新建一个,否则返回null。 + * @param id id可以为任意字符串,建议使用FromUserName作为id + * @param create + * @return + */ + WxSession getSession(String id, boolean create); + + /** + *
+   * 设置WxSessionManager,只有当需要使用个性化的WxSessionManager的时候才需要调用此方法,
+   * WxCpService默认使用的是{@link me.chanjar.weixin.common.session.StandardSessionManager}
+   * 
+ * @param sessionManager + */ + void setSessionManager(WxSessionManager 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 ac13edaa..f0da30b5 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,6 +12,9 @@ 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.StandardSessionManager; +import me.chanjar.weixin.common.session.WxSession; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.StringUtils; import me.chanjar.weixin.common.util.crypto.SHA1; import me.chanjar.weixin.common.util.fs.FileUtils; @@ -34,6 +37,8 @@ import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.impl.client.BasicResponseHandler; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -41,10 +46,11 @@ 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 { + protected final Logger log = LoggerFactory.getLogger(WxCpServiceImpl.class); + /** * 全局的是否正在刷新access token的锁 */ @@ -52,12 +58,16 @@ public class WxCpServiceImpl implements WxCpService { protected WxCpConfigStorage wxCpConfigStorage; - protected final ThreadLocal retryTimes = new ThreadLocal(); - protected CloseableHttpClient httpClient; protected HttpHost httpProxy; + private int retrySleepMillis = 1000; + + private int maxRetryTimes = 5; + + protected WxSessionManager sessionManager = new StandardSessionManager(); + public boolean checkSignature(String msgSignature, String timestamp, String nonce, String data) { try { return SHA1.gen(wxCpConfigStorage.getToken(), timestamp, nonce, data).equals(msgSignature); @@ -366,6 +376,33 @@ public class WxCpServiceImpl implements WxCpService { * @throws WxErrorException */ public T execute(RequestExecutor executor, String uri, E data) throws WxErrorException { + int retryTimes = 0; + do { + try { + return executeInternal(executor, uri, data); + } catch (WxErrorException e) { + WxError error = e.getError(); + /** + * -1 系统繁忙, 1000ms后重试 + */ + if (error.getErrorCode() == -1) { + int sleepMillis = retrySleepMillis * (1 << retryTimes); + try { + log.debug("微信系统繁忙,{}ms 后重试(第{}次)", sleepMillis, retryTimes + 1); + Thread.sleep(sleepMillis); + } catch (InterruptedException e1) { + throw new RuntimeException(e1); + } + } else { + throw e; + } + } + } while(++retryTimes < maxRetryTimes); + + throw new RuntimeException("微信服务端异常,超出重试次数"); + } + + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { String accessToken = getAccessToken(false); String uriWithAccessToken = uri; @@ -381,31 +418,10 @@ public class WxCpServiceImpl implements WxCpService { * 42001 access_token超时 */ if (error.getErrorCode() == 42001 || error.getErrorCode() == 40001) { - // 强制设置wxCpConfigStorage它的access token过期了,这样在下一次请求里就会刷新access token + // 强制设置wxMpConfigStorage它的access token过期了,这样在下一次请求里就会刷新access token wxCpConfigStorage.expireAccessToken(); return execute(executor, uri, data); } - /** - * -1 系统繁忙, 1000ms后重试 - */ - if (error.getErrorCode() == -1) { - if (retryTimes.get() == null) { - retryTimes.set(0); - } - if (retryTimes.get() > 4) { - retryTimes.set(0); - throw new RuntimeException("微信服务端异常,超出重试次数"); - } - int sleepMillis = 1000 * (1 << retryTimes.get()); - try { - System.out.println("微信系统繁忙," + sleepMillis + "ms后重试"); - Thread.sleep(sleepMillis); - retryTimes.set(retryTimes.get() + 1); - return execute(executor, uri, data); - } catch (InterruptedException e1) { - throw new RuntimeException(e1); - } - } if (error.getErrorCode() != 0) { throw new WxErrorException(error); } @@ -416,7 +432,6 @@ public class WxCpServiceImpl implements WxCpService { throw new RuntimeException(e); } } - protected CloseableHttpClient getHttpclient() { return httpClient; } @@ -451,4 +466,37 @@ public class WxCpServiceImpl implements WxCpService { } } + @Override + public void setRetrySleepMillis(int retrySleepMillis) { + this.retrySleepMillis = retrySleepMillis; + } + + + @Override + public void setMaxRetryTimes(int maxRetryTimes) { + this.maxRetryTimes = maxRetryTimes; + } + + @Override + public WxSession getSession(String id) { + if (sessionManager == null) { + return null; + } + return sessionManager.getSession(id); + } + + @Override + public WxSession getSession(String id, boolean create) { + if (sessionManager == null) { + return null; + } + return sessionManager.getSession(id, create); + } + + + @Override + public void setSessionManager(WxSessionManager sessionManager) { + this.sessionManager = sessionManager; + } + } diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/util/xml/XStreamTransformer.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/util/xml/XStreamTransformer.java index 13eeac17..c6c5a5f1 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/util/xml/XStreamTransformer.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/util/xml/XStreamTransformer.java @@ -8,9 +8,6 @@ import java.io.InputStream; import java.util.HashMap; import java.util.Map; -/** - * Created by qianjia on 15/1/19. - */ public class XStreamTransformer { protected static final Map CLASS_2_XSTREAM_INSTANCE = configXStreamInstance(); diff --git a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java new file mode 100644 index 00000000..cf129088 --- /dev/null +++ b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java @@ -0,0 +1,66 @@ +package me.chanjar.weixin.cp.api; + +import me.chanjar.weixin.common.bean.result.WxError; +import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.util.http.RequestExecutor; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +@Test +public class WxCpBusyRetryTest { + + @DataProvider(name="getService") + public Object[][] getService() { + WxCpService service = new WxCpServiceImpl() { + + @Override + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { + WxError error = new WxError(); + error.setErrorCode(-1); + throw new WxErrorException(error); + } + }; + + service.setMaxRetryTimes(3); + service.setRetrySleepMillis(500); + return new Object[][] { + new Object[] { service } + }; + } + + @Test(dataProvider = "getService", expectedExceptions = RuntimeException.class) + public void testRetry(WxCpService service) throws WxErrorException { + service.execute(null, null, null); + } + + @Test(dataProvider = "getService") + public void testRetryInThreadPool(final WxCpService service) throws InterruptedException, ExecutionException { + // 当线程池中的线程复用的时候,还是能保证相同的重试次数 + ExecutorService executorService = Executors.newFixedThreadPool(1); + Runnable runnable = new Runnable() { + @Override + public void run() { + try { + System.out.println("====================="); + System.out.println(Thread.currentThread().getName() + ": testRetry"); + service.execute(null, null, null); + } catch (WxErrorException e) { + throw new RuntimeException(e); + } catch (RuntimeException e) { + // OK + } + } + }; + Future submit1 = executorService.submit(runnable); + Future submit2 = executorService.submit(runnable); + + submit1.get(); + submit2.get(); + } + +} 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 b3dc6c30..7394bbfa 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,8 @@ package me.chanjar.weixin.cp.api; import me.chanjar.weixin.common.api.WxConsts; +import me.chanjar.weixin.common.session.StandardSessionManager; +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 +69,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,11 +152,145 @@ 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; } } + @DataProvider + public Object[][] standardSessionManager() { + + // 故意把session存活时间变短,清理更频繁 + StandardSessionManager ism = new StandardSessionManager(); + ism.setMaxInactiveInterval(1); + ism.setProcessExpiresFrequency(1); + ism.setBackgroundProcessorDelay(1); + + return new Object[][] { + new Object[] { ism } + }; + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean1(StandardSessionManager ism) throws InterruptedException { + + // 两个同步请求,看是否处理完毕后会被清理掉 + final WxCpMessageRouter router = new WxCpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(false).handler(new WxSessionMessageHandler()).next() + .rule().async(false).handler(new WxSessionMessageHandler()).end(); + + WxCpXmlMessage msg = new WxCpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean2(StandardSessionManager ism) throws InterruptedException { + + // 1个同步,1个异步请求,看是否处理完毕后会被清理掉 + { + final WxCpMessageRouter router = new WxCpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(false).handler(new WxSessionMessageHandler()).next() + .rule().async(true).handler(new WxSessionMessageHandler()).end(); + + WxCpXmlMessage msg = new WxCpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + { + final WxCpMessageRouter router = new WxCpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(true).handler(new WxSessionMessageHandler()).next() + .rule().async(false).handler(new WxSessionMessageHandler()).end(); + + WxCpXmlMessage msg = new WxCpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean3(StandardSessionManager ism) throws InterruptedException { + + // 2个异步请求,看是否处理完毕后会被清理掉 + final WxCpMessageRouter router = new WxCpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(true).handler(new WxSessionMessageHandler()).next() + .rule().async(true).handler(new WxSessionMessageHandler()).end(); + + WxCpXmlMessage msg = new WxCpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean4(StandardSessionManager ism) throws InterruptedException { + + // 一个同步请求,看是否处理完毕后会被清理掉 + { + final WxCpMessageRouter router = new WxCpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(false).handler(new WxSessionMessageHandler()).end(); + + WxCpXmlMessage msg = new WxCpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + + { + final WxCpMessageRouter router = new WxCpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(true).handler(new WxSessionMessageHandler()).end(); + + WxCpXmlMessage msg = new WxCpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + } + + public static class WxSessionMessageHandler implements WxCpMessageHandler { + + @Override + public WxCpXmlOutMessage handle(WxCpXmlMessage wxMessage, Map context, WxCpService wxCpService, + WxSessionManager sessionManager) { + sessionManager.getSession(wxMessage.getFromUserName()); + 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 81786100..0dadee91 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-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpOAuth2Servlet.java b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpOAuth2Servlet.java index 02b1bf5d..e0800061 100644 --- a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpOAuth2Servlet.java +++ b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/demo/WxCpOAuth2Servlet.java @@ -10,9 +10,6 @@ import javax.servlet.http.HttpServletResponse; import java.io.IOException; import java.util.Arrays; -/** - * Created by qianjia on 14/11/28. - */ public class WxCpOAuth2Servlet extends HttpServlet { protected WxCpService wxCpService; diff --git a/weixin-java-cp/src/test/resources/logback-test.xml b/weixin-java-cp/src/test/resources/logback-test.xml new file mode 100644 index 00000000..9c2ec6ae --- /dev/null +++ b/weixin-java-cp/src/test/resources/logback-test.xml @@ -0,0 +1,16 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + diff --git a/weixin-java-cp/src/test/resources/testng.xml b/weixin-java-cp/src/test/resources/testng.xml index 4cddddcf..ba92bc0c 100644 --- a/weixin-java-cp/src/test/resources/testng.xml +++ b/weixin-java-cp/src/test/resources/testng.xml @@ -1,28 +1,29 @@ - - - - - - - - - - - - + + + + + + + + + + + + + - - - - - - - - - - - + + + + + + + + + + + diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpInMemoryConfigStorage.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpInMemoryConfigStorage.java index c694dde8..ff272f77 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpInMemoryConfigStorage.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpInMemoryConfigStorage.java @@ -152,7 +152,6 @@ public class WxMpInMemoryConfigStorage implements WxMpConfigStorage { this.http_proxy_password = http_proxy_password; } - @Override public String toString() { return "WxMpInMemoryConfigStorage{" + @@ -166,6 +165,8 @@ public class WxMpInMemoryConfigStorage implements WxMpConfigStorage { ", http_proxy_port=" + http_proxy_port + ", http_proxy_username='" + http_proxy_username + '\'' + ", http_proxy_password='" + http_proxy_password + '\'' + + ", jsapiTicket='" + jsapiTicket + '\'' + + ", jsapiTicketExpiresTime='" + jsapiTicketExpiresTime + '\'' + '}'; } 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 8e6deeb6..d7c30a31 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 d940804f..2b68a6f2 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,23 +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 qianjia * + * @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 7fcc3c96..afd8d3fb 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,16 +1,22 @@ package me.chanjar.weixin.mp.api; -import me.chanjar.weixin.common.util.WxMsgIdDuplicateChecker; -import me.chanjar.weixin.common.util.WxMsgIdInMemoryDuplicateChecker; +import me.chanjar.weixin.common.api.WxConsts; +import me.chanjar.weixin.common.session.*; +import me.chanjar.weixin.common.util.WxMessageDuplicateChecker; +import me.chanjar.weixin.common.util.WxMessageInMemoryDuplicateChecker; import me.chanjar.weixin.mp.bean.WxMpXmlMessage; import me.chanjar.weixin.mp.bean.WxMpXmlOutMessage; +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; /** @@ -38,12 +44,12 @@ import java.util.regex.Pattern; * router.route(message); * * - * @author qianjia - * */ public class WxMpMessageRouter { - private static final int DEFAULT_THREAD_POOL_SIZE = 20; + protected final Logger log = LoggerFactory.getLogger(WxMpMessageRouter.class); + + private static final int DEFAULT_THREAD_POOL_SIZE = 100; private final List rules = new ArrayList(); @@ -51,22 +57,22 @@ public class WxMpMessageRouter { private ExecutorService executorService; - private WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker; + private WxMessageDuplicateChecker messageDuplicateChecker; + + private WxSessionManager sessionManager; public WxMpMessageRouter(WxMpService wxMpService) { this.wxMpService = wxMpService; this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE); - this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); - } - - public WxMpMessageRouter(WxMpService wxMpService, int threadPoolSize) { - this.wxMpService = wxMpService; - this.executorService = Executors.newFixedThreadPool(threadPoolSize); - this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); + this.messageDuplicateChecker = new WxMessageInMemoryDuplicateChecker(); + this.sessionManager = new StandardSessionManager(); } /** - * 设置自定义的ExecutorService + *
+   * 设置自定义的 {@link ExecutorService}
+   * 如果不调用该方法,默认使用 Executors.newFixedThreadPool(100)
+   * 
* @param executorService */ public void setExecutorService(ExecutorService executorService) { @@ -74,11 +80,25 @@ public class WxMpMessageRouter { } /** - * 设置自定义的WxMsgIdDuplicateChecker - * @param wxMsgIdDuplicateChecker + *
+   * 设置自定义的 {@link me.chanjar.weixin.common.util.WxMessageDuplicateChecker}
+   * 如果不调用该方法,默认使用 {@link me.chanjar.weixin.common.util.WxMessageInMemoryDuplicateChecker}
+   * 
+ * @param messageDuplicateChecker */ - public void setWxMsgIdDuplicateChecker(WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker) { - this.wxMsgIdDuplicateChecker = wxMsgIdDuplicateChecker; + 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; } /** @@ -86,7 +106,7 @@ public class WxMpMessageRouter { * @return */ public Rule rule() { - return new Rule(this, wxMpService); + return new Rule(this, wxMpService, sessionManager); } /** @@ -94,7 +114,7 @@ public class WxMpMessageRouter { * @param wxMessage */ public WxMpXmlOutMessage route(final WxMpXmlMessage wxMessage) { - if (wxMsgIdDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { + if (isDuplicateMessage(wxMessage)) { // 如果是重复消息,那么就不做处理 return null; } @@ -109,33 +129,90 @@ public class WxMpMessageRouter { } } } - + if (matchRules.size() == 0) { return null; } - + WxMpXmlOutMessage 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访问结束 + log.debug("End session access: async=false, sessionId={}", wxMessage.getFromUserName()); + sessionEndAccess(wxMessage); } } + + if (futures.size() > 0) { + executorService.submit(new Runnable() { + @Override + public void run() { + for (Future future : futures) { + try { + future.get(); + log.debug("End session access: async=true, sessionId={}", wxMessage.getFromUserName()); + // 异步操作结束,session访问结束 + sessionEndAccess(wxMessage); + } catch (InterruptedException e) { + log.error("Error happened when wait task finish", e); + } catch (ExecutionException e) { + log.error("Error happened when wait task finish", e); + } + } + } + }); + } return res; } - + + protected boolean isDuplicateMessage(WxMpXmlMessage wxMessage) { + + String messageId = ""; + if (wxMessage.getMsgId() == null) { + messageId = wxMessage.getFromUserName() + "-" + String.valueOf(wxMessage.getCreateTime()); + } else { + messageId = String.valueOf(wxMessage.getMsgId()); + } + + if (messageDuplicateChecker.isDuplicate(messageId)) { + return true; + } + return false; + + } + + /** + * 对session的访问结束 + * @param wxMessage + */ + protected void sessionEndAccess(WxMpXmlMessage wxMessage) { + + InternalSession session = ((InternalSessionManager)sessionManager).findSession(wxMessage.getFromUserName()); + if (session != null) { + session.endAccess(); + } + + } + public static class Rule { private final WxMpMessageRouter routerBuilder; private final WxMpService wxMpService; + private final WxSessionManager sessionManager; + private boolean async = true; private String fromUser; @@ -155,10 +232,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; } /** @@ -324,7 +402,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; } } @@ -333,7 +411,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/WxMpService.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java index 7b194ccd..6bd501f3 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java @@ -3,6 +3,8 @@ package me.chanjar.weixin.mp.api; import me.chanjar.weixin.common.bean.WxMenu; import me.chanjar.weixin.common.bean.result.WxMediaUploadResult; import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.session.WxSession; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.http.RequestExecutor; import me.chanjar.weixin.mp.bean.*; import me.chanjar.weixin.mp.bean.result.*; @@ -456,9 +458,28 @@ public interface WxMpService { */ public T execute(RequestExecutor executor, String uri, E data) throws WxErrorException; - /** - * 注入 {@link WxMpConfigStorage} 的实现 - * @param wxConfigProvider - */ + /** + * 注入 {@link WxMpConfigStorage} 的实现 + * @param wxConfigProvider + */ public void setWxMpConfigStorage(WxMpConfigStorage wxConfigProvider); + + /** + *
+   * 设置当微信系统响应系统繁忙时,要等待多少 retrySleepMillis(ms) * 2^(重试次数 - 1) 再发起重试
+   * 默认:1000ms
+   * 
+ * @param retrySleepMillis + */ + void setRetrySleepMillis(int retrySleepMillis); + + /** + *
+   * 设置当微信系统响应系统繁忙时,最大重试次数
+   * 默认:5次
+   * 
+ * @param maxRetryTimes + */ + void setMaxRetryTimes(int maxRetryTimes); + } 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 ba256865..439c44df 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,6 +10,8 @@ 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.StandardSessionManager; +import me.chanjar.weixin.common.session.WxSessionManager; import me.chanjar.weixin.common.util.StringUtils; import me.chanjar.weixin.common.util.crypto.SHA1; import me.chanjar.weixin.common.util.fs.FileUtils; @@ -31,6 +33,8 @@ import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.impl.client.BasicResponseHandler; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -42,6 +46,8 @@ import java.util.UUID; public class WxMpServiceImpl implements WxMpService { + protected final Logger log = LoggerFactory.getLogger(WxMpServiceImpl.class); + /** * 全局的是否正在刷新access token的锁 */ @@ -54,12 +60,16 @@ public class WxMpServiceImpl implements WxMpService { protected WxMpConfigStorage wxMpConfigStorage; - protected final ThreadLocal retryTimes = new ThreadLocal(); - protected CloseableHttpClient httpClient; protected HttpHost httpProxy; + private int retrySleepMillis = 1000; + + private int maxRetryTimes = 5; + + protected WxSessionManager sessionManager = new StandardSessionManager(); + public boolean checkSignature(String timestamp, String nonce, String signature) { try { return SHA1.gen(wxMpConfigStorage.getToken(), timestamp, nonce).equals(signature); @@ -439,7 +449,7 @@ public class WxMpServiceImpl implements WxMpService { return execute(new SimplePostRequestExecutor(), url, postData); } - /** + /** * 向微信端发送请求,在这里执行的策略是当发生access_token过期时才去刷新,然后重新执行请求,而不是全局定时请求 * @param executor * @param uri @@ -448,6 +458,33 @@ public class WxMpServiceImpl implements WxMpService { * @throws WxErrorException */ public T execute(RequestExecutor executor, String uri, E data) throws WxErrorException { + int retryTimes = 0; + do { + try { + return executeInternal(executor, uri, data); + } catch (WxErrorException e) { + WxError error = e.getError(); + /** + * -1 系统繁忙, 1000ms后重试 + */ + if (error.getErrorCode() == -1) { + int sleepMillis = retrySleepMillis * (1 << retryTimes); + try { + log.debug("微信系统繁忙,{}ms 后重试(第{}次)", sleepMillis, retryTimes + 1); + Thread.sleep(sleepMillis); + } catch (InterruptedException e1) { + throw new RuntimeException(e1); + } + } else { + throw e; + } + } + } while(++retryTimes < maxRetryTimes); + + throw new RuntimeException("微信服务端异常,超出重试次数"); + } + + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { String accessToken = getAccessToken(false); String uriWithAccessToken = uri; @@ -467,27 +504,6 @@ public class WxMpServiceImpl implements WxMpService { wxMpConfigStorage.expireAccessToken(); return execute(executor, uri, data); } - /** - * -1 系统繁忙, 1000ms后重试 - */ - if (error.getErrorCode() == -1) { - if(retryTimes.get() == null) { - retryTimes.set(0); - } - if (retryTimes.get() > 4) { - retryTimes.set(0); - throw new RuntimeException("微信服务端异常,超出重试次数"); - } - int sleepMillis = 1000 * (1 << retryTimes.get()); - try { - System.out.println("微信系统繁忙," + sleepMillis + "ms后重试"); - Thread.sleep(sleepMillis); - retryTimes.set(retryTimes.get() + 1); - return execute(executor, uri, data); - } catch (InterruptedException e1) { - throw new RuntimeException(e1); - } - } if (error.getErrorCode() != 0) { throw new WxErrorException(error); } @@ -533,4 +549,15 @@ public class WxMpServiceImpl implements WxMpService { } } + @Override + public void setRetrySleepMillis(int retrySleepMillis) { + this.retrySleepMillis = retrySleepMillis; + } + + + @Override + public void setMaxRetryTimes(int maxRetryTimes) { + this.maxRetryTimes = maxRetryTimes; + } + } diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/WxMpTemplateMessage.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/WxMpTemplateMessage.java index cd6ac5f7..91441cd1 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/WxMpTemplateMessage.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/WxMpTemplateMessage.java @@ -5,9 +5,6 @@ import me.chanjar.weixin.mp.util.json.WxMpGsonBuilder; import java.util.ArrayList; import java.util.List; -/** - * Created by qianjia on 14/11/1. - */ public class WxMpTemplateMessage { private String toUser; diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/result/WxMpOAuth2AccessToken.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/result/WxMpOAuth2AccessToken.java index d2a6ef8e..e2d0d1a0 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/result/WxMpOAuth2AccessToken.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/bean/result/WxMpOAuth2AccessToken.java @@ -2,9 +2,6 @@ package me.chanjar.weixin.mp.bean.result; import me.chanjar.weixin.mp.util.json.WxMpGsonBuilder; -/** - * Created by qianjia on 14/11/26. - */ public class WxMpOAuth2AccessToken { private String accessToken; diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpCustomMessageGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpCustomMessageGsonAdapter.java index 488dd8df..3dc0515d 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpCustomMessageGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpCustomMessageGsonAdapter.java @@ -14,11 +14,6 @@ import me.chanjar.weixin.mp.bean.WxMpCustomMessage; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxMpCustomMessageGsonAdapter implements JsonSerializer { public JsonElement serialize(WxMpCustomMessage message, Type typeOfSrc, JsonSerializationContext context) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpGroupGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpGroupGsonAdapter.java index 63f0624e..58e6f98d 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpGroupGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpGroupGsonAdapter.java @@ -14,11 +14,6 @@ import me.chanjar.weixin.mp.bean.WxMpGroup; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxMpGroupGsonAdapter implements JsonSerializer, JsonDeserializer { public JsonElement serialize(WxMpGroup group, Type typeOfSrc, JsonSerializationContext context) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassGroupMessageGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassGroupMessageGsonAdapter.java index a0077469..79a9a0c6 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassGroupMessageGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassGroupMessageGsonAdapter.java @@ -17,11 +17,6 @@ import me.chanjar.weixin.mp.bean.WxMpMassGroupMessage; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxMpMassGroupMessageGsonAdapter implements JsonSerializer { public JsonElement serialize(WxMpMassGroupMessage message, Type typeOfSrc, JsonSerializationContext context) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassNewsGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassNewsGsonAdapter.java index 5ba78594..1719b75a 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassNewsGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassNewsGsonAdapter.java @@ -13,11 +13,6 @@ import me.chanjar.weixin.mp.bean.WxMpMassNews; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxMpMassNewsGsonAdapter implements JsonSerializer { public JsonElement serialize(WxMpMassNews message, Type typeOfSrc, JsonSerializationContext context) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassOpenIdsMessageGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassOpenIdsMessageGsonAdapter.java index cd03addf..84063da4 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassOpenIdsMessageGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpMassOpenIdsMessageGsonAdapter.java @@ -14,11 +14,6 @@ import me.chanjar.weixin.mp.bean.WxMpMassOpenIdsMessage; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxMpMassOpenIdsMessageGsonAdapter implements JsonSerializer { public JsonElement serialize(WxMpMassOpenIdsMessage message, Type typeOfSrc, JsonSerializationContext context) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpOAuth2AccessTokenAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpOAuth2AccessTokenAdapter.java index 0c3d4acd..6d73131a 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpOAuth2AccessTokenAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpOAuth2AccessTokenAdapter.java @@ -6,9 +6,6 @@ import me.chanjar.weixin.mp.bean.result.WxMpOAuth2AccessToken; import java.lang.reflect.Type; -/** - * Created by qianjia on 14/11/28. - */ public class WxMpOAuth2AccessTokenAdapter implements JsonDeserializer { public WxMpOAuth2AccessToken deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpTemplateMessageGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpTemplateMessageGsonAdapter.java index 21195b4d..c34ac2c2 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpTemplateMessageGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxMpTemplateMessageGsonAdapter.java @@ -17,9 +17,6 @@ import me.chanjar.weixin.mp.bean.WxMpTemplateMessage; import java.lang.reflect.Type; -/** - * @author qianjia - */ public class WxMpTemplateMessageGsonAdapter implements JsonSerializer { public JsonElement serialize(WxMpTemplateMessage message, Type typeOfSrc, JsonSerializationContext context) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserGsonAdapter.java index cbbc3080..462994f6 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserGsonAdapter.java @@ -14,11 +14,6 @@ import me.chanjar.weixin.mp.bean.result.WxMpUser; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxUserGsonAdapter implements JsonDeserializer { public WxMpUser deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserListGsonAdapter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserListGsonAdapter.java index ec65735e..5be3d160 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserListGsonAdapter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/json/WxUserListGsonAdapter.java @@ -14,11 +14,6 @@ import me.chanjar.weixin.mp.bean.result.WxMpUserList; import java.lang.reflect.Type; -/** - * - * @author qianjia - * - */ public class WxUserListGsonAdapter implements JsonDeserializer { public WxMpUserList deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/xml/XStreamTransformer.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/xml/XStreamTransformer.java index 4699b01d..7b20c6f1 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/xml/XStreamTransformer.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/util/xml/XStreamTransformer.java @@ -8,9 +8,6 @@ import java.io.InputStream; import java.util.HashMap; import java.util.Map; -/** - * Created by qianjia on 15/1/19. - */ public class XStreamTransformer { protected static final Map CLASS_2_XSTREAM_INSTANCE = configXStreamInstance(); diff --git a/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java new file mode 100644 index 00000000..2e16723a --- /dev/null +++ b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java @@ -0,0 +1,66 @@ +package me.chanjar.weixin.mp.api; + +import me.chanjar.weixin.common.bean.result.WxError; +import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.util.http.RequestExecutor; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +@Test +public class WxMpBusyRetryTest { + + @DataProvider(name="getService") + public Object[][] getService() { + WxMpService service = new WxMpServiceImpl() { + + @Override + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { + WxError error = new WxError(); + error.setErrorCode(-1); + throw new WxErrorException(error); + } + }; + + service.setMaxRetryTimes(3); + service.setRetrySleepMillis(500); + return new Object[][] { + new Object[] { service } + }; + } + + @Test(dataProvider = "getService", expectedExceptions = RuntimeException.class) + public void testRetry(WxMpService service) throws WxErrorException { + service.execute(null, null, null); + } + + @Test(dataProvider = "getService") + public void testRetryInThreadPool(final WxMpService service) throws InterruptedException, ExecutionException { + // 当线程池中的线程复用的时候,还是能保证相同的重试次数 + ExecutorService executorService = Executors.newFixedThreadPool(1); + Runnable runnable = new Runnable() { + @Override + public void run() { + try { + System.out.println("====================="); + System.out.println(Thread.currentThread().getName() + ": testRetry"); + service.execute(null, null, null); + } catch (WxErrorException e) { + throw new RuntimeException(e); + } catch (RuntimeException e) { + // OK + } + } + }; + Future submit1 = executorService.submit(runnable); + Future submit2 = executorService.submit(runnable); + + submit1.get(); + submit2.get(); + } + +} 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 3f7e841e..0b636985 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,8 @@ package me.chanjar.weixin.mp.api; import me.chanjar.weixin.common.api.WxConsts; +import me.chanjar.weixin.common.session.StandardSessionManager; +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 +69,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,11 +152,144 @@ 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; } } + @DataProvider + public Object[][] standardSessionManager() { + + // 故意把session存活时间变短,清理更频繁 + StandardSessionManager ism = new StandardSessionManager(); + ism.setMaxInactiveInterval(1); + ism.setProcessExpiresFrequency(1); + ism.setBackgroundProcessorDelay(1); + + return new Object[][] { + new Object[] { ism } + }; + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean1(StandardSessionManager ism) throws InterruptedException { + + // 两个同步请求,看是否处理完毕后会被清理掉 + final WxMpMessageRouter router = new WxMpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(false).handler(new WxSessionMessageHandler()).next() + .rule().async(false).handler(new WxSessionMessageHandler()).end(); + + WxMpXmlMessage msg = new WxMpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean2(StandardSessionManager ism) throws InterruptedException { + + // 1个同步,1个异步请求,看是否处理完毕后会被清理掉 + { + final WxMpMessageRouter router = new WxMpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(false).handler(new WxSessionMessageHandler()).next() + .rule().async(true).handler(new WxSessionMessageHandler()).end(); + + WxMpXmlMessage msg = new WxMpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + { + final WxMpMessageRouter router = new WxMpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(true).handler(new WxSessionMessageHandler()).next() + .rule().async(false).handler(new WxSessionMessageHandler()).end(); + + WxMpXmlMessage msg = new WxMpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean3(StandardSessionManager ism) throws InterruptedException { + + // 2个异步请求,看是否处理完毕后会被清理掉 + final WxMpMessageRouter router = new WxMpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(true).handler(new WxSessionMessageHandler()).next() + .rule().async(true).handler(new WxSessionMessageHandler()).end(); + + WxMpXmlMessage msg = new WxMpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + + } + + @Test(dataProvider = "standardSessionManager") + public void testSessionClean4(StandardSessionManager ism) throws InterruptedException { + + // 一个同步请求,看是否处理完毕后会被清理掉 + { + final WxMpMessageRouter router = new WxMpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(false).handler(new WxSessionMessageHandler()).end(); + + WxMpXmlMessage msg = new WxMpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + + { + final WxMpMessageRouter router = new WxMpMessageRouter(null); + router.setSessionManager(ism); + router + .rule().async(true).handler(new WxSessionMessageHandler()).end(); + + WxMpXmlMessage msg = new WxMpXmlMessage(); + msg.setFromUserName("abc"); + router.route(msg); + + Thread.sleep(2000l); + Assert.assertEquals(ism.getActiveSessions(), 0); + } + } + + public static class WxSessionMessageHandler implements WxMpMessageHandler { + + @Override + public WxMpXmlOutMessage handle(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService, + WxSessionManager sessionManager) { + sessionManager.getSession(wxMessage.getFromUserName()); + 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 c58671cf..8e44d2db 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 diff --git a/weixin-java-mp/src/test/resources/logback-test.xml b/weixin-java-mp/src/test/resources/logback-test.xml new file mode 100644 index 00000000..ed595101 --- /dev/null +++ b/weixin-java-mp/src/test/resources/logback-test.xml @@ -0,0 +1,16 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + diff --git a/weixin-java-mp/src/test/resources/testng.xml b/weixin-java-mp/src/test/resources/testng.xml index 333400e4..cf9c2d64 100644 --- a/weixin-java-mp/src/test/resources/testng.xml +++ b/weixin-java-mp/src/test/resources/testng.xml @@ -3,6 +3,7 @@ +