Browse Source

[线程][修改]:消息处理整理,取消异步调用

release
Stormeye Wu 7 years ago
parent
commit
629009be87
1 changed files with 235 additions and 126 deletions
  1. +235
    -126
      mlWechatOpen/src/main/java/com/iformall/controller/WechatCalllbackController.java

+ 235
- 126
mlWechatOpen/src/main/java/com/iformall/controller/WechatCalllbackController.java View File

@@ -12,10 +12,10 @@ import com.iformall.service.wechat.FmOpenService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import me.chanjar.weixin.common.error.WxErrorException;
import me.chanjar.weixin.mp.api.WxMpService;
import me.chanjar.weixin.mp.bean.kefu.WxMpKefuMessage;
import me.chanjar.weixin.mp.bean.message.WxMpXmlMessage;
import me.chanjar.weixin.mp.bean.message.WxMpXmlOutMessage;
import me.chanjar.weixin.mp.bean.result.WxMpUser;
import me.chanjar.weixin.open.bean.message.WxOpenXmlMessage;
import me.chanjar.weixin.open.bean.result.WxOpenAuthorizerInfoResult;
import me.chanjar.weixin.open.bean.result.WxOpenPlatformResult;
@@ -62,6 +62,9 @@ public class WechatCalllbackController extends BaseController {
@Autowired
private WxAppinfoService appinfoService;

@Autowired
private WxCUserService userService;

@Autowired
private MailService mailService;

@@ -85,27 +88,27 @@ public class WechatCalllbackController extends BaseController {
throw new IllegalArgumentException("非法请求,可能属于伪造的请求!");
}

InputStream inStream = request.getInputStream();
ByteArrayOutputStream outSteam = new ByteArrayOutputStream();
byte[] buffer = new byte[1024];
int len = 0;
while ((len = inStream.read(buffer)) != -1) {
outSteam.write(buffer, 0, len);
}
String resultxml = new String(outSteam.toByteArray(), Charset.forName("UTF-8"));
logger.info(resultxml);

outSteam.close();
inStream.close();
String resultxml = getRequestXML(request);

logger.info("\n接收微信回调:[signature=[{}], encType=[{}], msgSignature=[{}],"
+ " timestamp=[{}], nonce=[{}], requestBody=[\n{}\n] ",
signature, encType, msgSignature, timestamp, nonce, resultxml);

// aes加密的消息
// aes加密的消息, 解密
WxOpenXmlMessage inMessage = WxOpenXmlMessage.fromEncryptedXml(resultxml,
openService.getWxOpenConfigStorage(), timestamp, nonce, msgSignature);
this.logger.info("接收微信回调,消息解密后内容为:{} ", inMessage.toString());

// 处理
return handleTicket(inMessage);
}

/**
* 处理ticket消息
* @param inMessage
* @return
*/
private String handleTicket(WxOpenXmlMessage inMessage) {
if (inMessage.getInfoType().equals(EnumWxAuthorizationInfoType.COMPONENT_VERIFY_TICKET.getMessage())) {
// 保存component_verify_ticket
componentVerifyTicketService.saveOrUpdate(new WxComponentVerifyTicket() {
@@ -116,33 +119,11 @@ public class WechatCalllbackController extends BaseController {
setDeadline(new Date(Long.valueOf(System.currentTimeMillis() + (60 * 60 * 1000))));
}
});
return "success";
} else if (inMessage.getInfoType().equals(EnumWxAuthorizationInfoType.AUTHORIZED.getMessage()) ||
inMessage.getInfoType().equals(EnumWxAuthorizationInfoType.UPDATEAUTHORIZED.getMessage())) {
// 审核通过
try {
WxOpenQueryAuthResult queryAuthResult = openService.getWxOpenComponentService().getQueryAuth(inMessage.getAuthorizationCode());
logger.info("getQueryAuth", queryAuthResult);

WxOpenAuthorizerInfoResult openAuthorizerInfoResult = openService.getWxOpenComponentService().getAuthorizerInfo(
queryAuthResult.getAuthorizationInfo().getAuthorizerAppid());
logger.info(openAuthorizerInfoResult.toString());
// save auth info
authorizerInfoService.saveOrUpdate(new WxAuthorizerInfo() {
{
setAuthorizerAppid(openAuthorizerInfoResult.getAuthorizationInfo().getAuthorizerAppid());
setAlias(openAuthorizerInfoResult.getAuthorizerInfo().getAlias());
setHeadImg(openAuthorizerInfoResult.getAuthorizerInfo().getHeadImg());
setQrcodeUrl(openAuthorizerInfoResult.getAuthorizerInfo().getQrcodeUrl());
setRefreshToken(openAuthorizerInfoResult.getAuthorizationInfo().getAuthorizerRefreshToken());
setAccessToken(openAuthorizerInfoResult.getAuthorizationInfo().getAuthorizerAccessToken());
setAccessTokenExpire(new Date(Long.valueOf(System.currentTimeMillis() + openAuthorizerInfoResult.getAuthorizationInfo().getExpiresIn()*1000)));
setAuthorizationStatus(EnumWxAuthorizationStatus.AUTHORIZED.getCode());
setAuthTime(new Date());
}
});
} catch (WxErrorException e) {
logger.error(e.getMessage());
}
if (handleWeappAuditedMessage(inMessage)) return "success";
} else if (inMessage.getInfoType().equals(EnumWxAuthorizationInfoType.UNAUTHORIZED.getMessage())) {
// 授权取消
authorizerInfoService.saveOrUpdate(new WxAuthorizerInfo() {
@@ -152,6 +133,7 @@ public class WechatCalllbackController extends BaseController {
setAuthTime(new Date());
}
});
return "success";
}

try {
@@ -160,10 +142,44 @@ public class WechatCalllbackController extends BaseController {
} catch (WxErrorException e) {
this.logger.error("notify", e);
}

return "success";
}

/**
* 处理小程序审核消息
* @param inMessage
* @return
*/
private boolean handleWeappAuditedMessage(WxOpenXmlMessage inMessage) {
// 审核通过
try {
WxOpenQueryAuthResult queryAuthResult = openService.getWxOpenComponentService().getQueryAuth(inMessage.getAuthorizationCode());
logger.info("getQueryAuth", queryAuthResult);

WxOpenAuthorizerInfoResult openAuthorizerInfoResult = openService.getWxOpenComponentService().getAuthorizerInfo(
queryAuthResult.getAuthorizationInfo().getAuthorizerAppid());
logger.info(openAuthorizerInfoResult.toString());
// save auth info
authorizerInfoService.saveOrUpdate(new WxAuthorizerInfo() {
{
setAuthorizerAppid(openAuthorizerInfoResult.getAuthorizationInfo().getAuthorizerAppid());
setAlias(openAuthorizerInfoResult.getAuthorizerInfo().getAlias());
setHeadImg(openAuthorizerInfoResult.getAuthorizerInfo().getHeadImg());
setQrcodeUrl(openAuthorizerInfoResult.getAuthorizerInfo().getQrcodeUrl());
setRefreshToken(openAuthorizerInfoResult.getAuthorizationInfo().getAuthorizerRefreshToken());
setAccessToken(openAuthorizerInfoResult.getAuthorizationInfo().getAuthorizerAccessToken());
setAccessTokenExpire(new Date(Long.valueOf(System.currentTimeMillis() + openAuthorizerInfoResult.getAuthorizationInfo().getExpiresIn()*1000)));
setAuthorizationStatus(EnumWxAuthorizationStatus.AUTHORIZED.getCode());
setAuthTime(new Date());
}
});
return true;
} catch (WxErrorException e) {
logger.error(e.getMessage());
}
return false;
}

@RequestMapping(value = "/{appId}/callback")
@ApiOperation("消息回调")
public void callback(@PathVariable("appId") String appId, HttpServletRequest request, HttpServletResponse response) throws IOException {
@@ -181,18 +197,7 @@ public class WechatCalllbackController extends BaseController {
throw new IllegalArgumentException("非法请求,可能属于伪造的请求!");
}

InputStream inStream = request.getInputStream();
ByteArrayOutputStream outSteam = new ByteArrayOutputStream();
byte[] buffer = new byte[1024];
int len = 0;
while ((len = inStream.read(buffer)) != -1) {
outSteam.write(buffer, 0, len);
}
String resultxml = new String(outSteam.toByteArray(), Charset.forName("UTF-8"));
logger.info(resultxml);

outSteam.close();
inStream.close();
String resultxml = getRequestXML(request);

logger.info("接收微信APP回调:[appId=[{}], openid=[{}], signature=[{}], encType=[{}], msgSignature=[{}],"
+ " timestamp=[{}], nonce=[{}], requestBody=[\n{}\n] ",
@@ -200,101 +205,55 @@ public class WechatCalllbackController extends BaseController {


String out = "";
// aes加密的消息
// aes加密的消息, 解密
WxMpXmlMessage inMessage = WxOpenXmlMessage.fromEncryptedMpXml(resultxml,
openService.getWxOpenConfigStorage(), timestamp, nonce, msgSignature);
logger.info("接收微信APP回调,消息解密后内容为:\n{}", inMessage.toString());

// 全网发布测试用例
if (StringUtils.equalsAnyIgnoreCase(appId, "wx570bc396a51b8ff8", "wxd101a85aa106f53e")) {
try {
if (StringUtils.equals(inMessage.getMsgType(), "text")) {
if (StringUtils.equals(inMessage.getContent(), "TESTCOMPONENT_MSG_TYPE_TEXT")) {
out = WxOpenXmlMessage.wxMpOutXmlMessageToEncryptedXml(
WxMpXmlOutMessage.TEXT().content("TESTCOMPONENT_MSG_TYPE_TEXT_callback")
.fromUser(inMessage.getToUser())
.toUser(inMessage.getFromUser())
.build(),
openService.getWxOpenConfigStorage()
);
} else if (StringUtils.startsWith(inMessage.getContent(), "QUERY_AUTH_CODE:")) {
String msg = inMessage.getContent().replace("QUERY_AUTH_CODE:", "") + "_from_api";
WxMpKefuMessage kefuMessage = WxMpKefuMessage.TEXT().content(msg).toUser(inMessage.getFromUser()).build();
openService.getWxOpenComponentService().getWxMpServiceByAppid(appId).getKefuService().sendKefuMessage(kefuMessage);
}
} else if (StringUtils.equals(inMessage.getMsgType(), "event")) {
WxMpKefuMessage kefuMessage = WxMpKefuMessage.TEXT().content(inMessage.getEvent() + "from_callback").toUser(inMessage.getFromUser()).build();
openService.getWxOpenComponentService().getWxMpServiceByAppid(appId).getKefuService().sendKefuMessage(kefuMessage);
}
} catch (WxErrorException e) {
logger.error("callback", e);
}

handleWechatTest(appId, response, inMessage);
PrintWriter printWriter = response.getWriter();
printWriter.print("success");
printWriter.flush();
printWriter.close();
wechatWriteSuccess(printWriter);
return;
}

// 实际 消息 处理
handleCallbackMessage(appId, response, inMessage);
}

/**
* 实际回调消息处理
* @param appId
* @param response
* @param inMessage
* @throws IOException
*/
private void handleCallbackMessage(String appId,
HttpServletResponse response,
WxMpXmlMessage inMessage) throws IOException {
WxAuthorizerInfo authorizerInfo = authorizerInfoService.getByAppId(appId);

PrintWriter printWriter = response.getWriter();

if(authorizerInfo != null &&
(authorizerInfo.getType().equals(EnumAppType.B.getCode()) ||
authorizerInfo.getType().equals(EnumAppType.C.getCode()))) {
// 小程序 审核 消息
// 小程序 消息 处理
if(authorizerInfo != null && (authorizerInfo.getType() <= EnumAppType.C.getCode())) {
/**判断消息类型,调用对应的方法*/
switch (inMessage.getMsgType()) {
case "event": {
switch (inMessage.getEvent()) {
case "weapp_audit_success": { // 审核成功
updateAuditStatus(appId, true, "");
break;
}
case "weapp_audit_fail": { // 审核失败
updateAuditStatus(appId, false, inMessage.getReason());
break;
}
default:
break;
}
printWriter.print("success");
printWriter.flush();
printWriter.close();
return;
}
case "text":
WxMpXmlOutMessage outMessage = openService.getWxOpenMessageRouter().route(inMessage, appId);
if (outMessage == null) {
printWriter.print("success");
printWriter.flush();
printWriter.close();
return;
}
// out = WxOpenXmlMessage.wxMpOutXmlMessageToEncryptedXml(outMessage, openService.getWxOpenConfigStorage());
// logger.info(out);
break;
default:
printWriter.print("success");
printWriter.flush();
printWriter.close();
return;
}
handleWeappCallbackMessage(appId, inMessage);
wechatWriteSuccess(printWriter);
return;
} else {
logger.info("unknown event, appId: " + appId);

getThreadCount( 0 );
handleMPCallbackMessage(appId, inMessage, authorizerInfo);
wechatWriteSuccess(printWriter);
return;
/*
// 公众号
try {
WxMpXmlOutMessage outMessage = mpManager.getRouter(appId).route(inMessage);
getThreadCount( 1 );
if (outMessage == null) {
this.logger.info("2 return: success");
printWriter.print("success");
printWriter.flush();
printWriter.close();
wechatWriteSuccess(printWriter);
return;
}
// logger.info("组装回复" + outMessage.toXml());
@@ -303,13 +262,165 @@ public class WechatCalllbackController extends BaseController {
} catch (Exception e) {
logger.error("路由消息时出现异常!", e);
}
*/
}
}

/**
* 处理小程序回调消息
* @param appId
* @param inMessage
*/
private void handleWeappCallbackMessage(String appId, WxMpXmlMessage inMessage) {
switch (inMessage.getMsgType()) {
case "event": {
switch (inMessage.getEvent()) {
case "weapp_audit_success": { // 审核成功
updateAuditStatus(appId, true, "");
break;
}
case "weapp_audit_fail": { // 审核失败
updateAuditStatus(appId, false, inMessage.getReason());
break;
}
default:
break;
}
return;
}
default:
return;
}
}

/**
* 处理公众号回调消息
* @param appId
* @param inMessage
* @param authorizerInfo
*/
private void handleMPCallbackMessage(String appId, WxMpXmlMessage inMessage, WxAuthorizerInfo authorizerInfo) {
switch (inMessage.getMsgType()) {
case "event": {
switch (inMessage.getEvent()) {
case "subscribe": { // 关注
handleEventSubscribe(appId, inMessage, authorizerInfo);
break;
}
case "unsubscribe": { // 取消关注
handleEventUnsubscribe(appId, inMessage, authorizerInfo);
break;
}
default: {
break;
}
}
return;
}
default: {
return;
}
}
}

private void wechatWriteSuccess(PrintWriter printWriter) {
printWriter.print("success");
printWriter.flush();
printWriter.close();
}

/**
* 新关注消息处理
* @param appId
* @param inMessage
* @param authorizerInfo
*/
private void handleEventSubscribe(String appId, WxMpXmlMessage inMessage, WxAuthorizerInfo authorizerInfo) {
this.logger.info("新关注用户 OPENID: " + inMessage.getFromUser() + ", AppId: " + appId);

// 获取微信用户基本信息
try {
WxMpUser userWxInfo = openService.getWxOpenComponentService().getWxMpServiceByAppid(appId).getUserService()
.userInfo(inMessage.getFromUser());
if (userWxInfo != null) {
this.logger.info(userWxInfo.toString());
userService.saveOrUpdateMpUser(userWxInfo, authorizerInfo);
}
} catch (WxErrorException e) {
if (e.getError().getErrorCode() == 48001) {
this.logger.info("该公众号没有获取用户信息权限!");
}
}
}


/**
* 取消关注消息处理
* @param appId
* @param inMessage
* @param authorizerInfo
*/
private void handleEventUnsubscribe(String appId, WxMpXmlMessage inMessage, WxAuthorizerInfo authorizerInfo) {
String openId = inMessage.getFromUser();
this.logger.info("取消关注用户 OPENID: " + openId + ", AppId: " + appId);

WxMpUser mpUser = new WxMpUser();
mpUser.setOpenId(inMessage.getFromUser());
mpUser.setSubscribe(false);
mpUser.setSubscribeTime(inMessage.getCreateTime());

userService.saveOrUpdateMpUser(mpUser, authorizerInfo);
}

/**
* 测试账号特殊处理
* @param appId
* @param response
* @param inMessage
* @throws IOException
*/
private void handleWechatTest(@PathVariable("appId") String appId, HttpServletResponse response, WxMpXmlMessage inMessage) throws IOException {
String out;
try {
if (StringUtils.equals(inMessage.getMsgType(), "text")) {
if (StringUtils.equals(inMessage.getContent(), "TESTCOMPONENT_MSG_TYPE_TEXT")) {
out = WxOpenXmlMessage.wxMpOutXmlMessageToEncryptedXml(
WxMpXmlOutMessage.TEXT().content("TESTCOMPONENT_MSG_TYPE_TEXT_callback")
.fromUser(inMessage.getToUser())
.toUser(inMessage.getFromUser())
.build(),
openService.getWxOpenConfigStorage()
);
} else if (StringUtils.startsWith(inMessage.getContent(), "QUERY_AUTH_CODE:")) {
String msg = inMessage.getContent().replace("QUERY_AUTH_CODE:", "") + "_from_api";
WxMpKefuMessage kefuMessage = WxMpKefuMessage.TEXT().content(msg).toUser(inMessage.getFromUser()).build();
openService.getWxOpenComponentService().getWxMpServiceByAppid(appId).getKefuService().sendKefuMessage(kefuMessage);
}
} else if (StringUtils.equals(inMessage.getMsgType(), "event")) {
WxMpKefuMessage kefuMessage = WxMpKefuMessage.TEXT().content(inMessage.getEvent() + "from_callback").toUser(inMessage.getFromUser()).build();
openService.getWxOpenComponentService().getWxMpServiceByAppid(appId).getKefuService().sendKefuMessage(kefuMessage);
}
} catch (WxErrorException e) {
logger.error("callback", e);
}
}

private String getRequestXML(HttpServletRequest request) throws IOException {
InputStream inStream = request.getInputStream();
ByteArrayOutputStream outSteam = new ByteArrayOutputStream();
byte[] buffer = new byte[1024];
int len = 0;
while ((len = inStream.read(buffer)) != -1) {
outSteam.write(buffer, 0, len);
}
String resultxml = new String(outSteam.toByteArray(), Charset.forName("UTF-8"));
logger.info(resultxml);

outSteam.close();
inStream.close();
return resultxml;
}

private void getThreadCount( int i) {
ThreadGroup group = Thread.currentThread().getThreadGroup();
ThreadGroup topGroup = group;
@@ -326,8 +437,6 @@ public class WechatCalllbackController extends BaseController {
}

private void updateAuditStatus(String appId, boolean success, String reason) {


WxWeappAuditStatus auditStatus = weappAuditStatusService.getByAppId(appId);
if (auditStatus != null) {
if (success) {


Loading…
Cancel
Save