228 lines
9.4 KiB
C#
228 lines
9.4 KiB
C#
using DAL.Redis;
|
||
using Newtonsoft.Json;
|
||
using Quartz;
|
||
using Quartz.Xml;
|
||
using SuperSocket.WebSocket;
|
||
using System;
|
||
using System.Collections;
|
||
using System.Collections.Generic;
|
||
using System.IO;
|
||
using System.ServiceProcess;
|
||
using System.Threading;
|
||
using System.Threading.Tasks;
|
||
using WeChatServerServices.Model;
|
||
using WX.CRM.Common;
|
||
using WX.CRM.DAL.Redis;
|
||
|
||
namespace WeChatServerServices
|
||
{
|
||
public partial class WeChatServerServices : ServiceBase
|
||
{
|
||
private RedisSet<string> _redisSet = new RedisSet<string>();
|
||
//private RedisList<string> _redisList = new RedisList<string>();
|
||
//private RedisList<rcontact> _redisList = new RedisList<rcontact>();
|
||
//private RedisList<message> _redisList2 = new RedisList<message>();
|
||
private readonly PubSub _sub = new PubSub();
|
||
private static Dictionary<string, int> _sessionList = new Dictionary<string, int>();
|
||
private static WebSocketServer _appServer = null;
|
||
//public static string _shareName;
|
||
//private static string _isShare;
|
||
//private static string _deviceName;
|
||
|
||
static WeChatServerServices()
|
||
{
|
||
//用户名勿比指定共享服务器的IP或名称，否则会引起1312错误
|
||
//_isShare = Utility.GetSettingByKey("IsShare");
|
||
//if (!string.IsNullOrEmpty(_isShare))
|
||
//{
|
||
// _shareName = Utility.GetSettingByKey("NetUseShareName") ?? @"\\192.168.1.171\weixin";
|
||
// _deviceName = Utility.GetSettingByKey("NetUseDeviceName") ?? "z:";
|
||
// string user = Utility.GetSettingByKey("NetUseUser") ?? @"192.168.1.171\admin";
|
||
// string pwd = Utility.GetSettingByKey("NetUsePwd") ?? "read,./1";
|
||
// NetUseHelper.Build(_shareName, user, pwd, string.Empty);//不指定盘符，避免引起盘符被占用的错误
|
||
//}
|
||
}
|
||
public WeChatServerServices()
|
||
{
|
||
InitializeComponent();
|
||
//if (_appServer == null)
|
||
//{
|
||
// _appServer = new WebSocketServer();
|
||
// //_appServer.NewSessionConnected += null;
|
||
// //_appServer.SessionClosed += null;
|
||
// //_appServer.NewMessageReceived += null;
|
||
// //_appServer.NewDataReceived += null;
|
||
// _appServer.NewSessionConnected += _appServer_NewSessionConnected; ;
|
||
// _appServer.SessionClosed += _appServer_SessionClosed; ;
|
||
// _appServer.NewMessageReceived += _appServer_NewMessageReceived; ;
|
||
// _appServer.NewDataReceived += _appServer_NewDataReceived; ;
|
||
//}
|
||
|
||
}
|
||
|
||
private void _appServer_NewDataReceived(WebSocketSession session, byte[] value)
|
||
{
|
||
throw new NotImplementedException();
|
||
}
|
||
|
||
private void _appServer_NewMessageReceived(WebSocketSession session, string value)
|
||
{
|
||
var msg = string.Format("接收到客户端:{0}，消息:{1}", session.SessionID, value);
|
||
session.Logger.Info(msg);
|
||
//var sessionId = _sessionList.FirstOrDefault(p => p.Value == 1).Key;
|
||
//SendMsgToRemotePoint(sessionId, value);
|
||
if (value.StartsWith("rcontact:"))
|
||
{
|
||
var index = value.IndexOf(':');
|
||
var json = value.Substring(index + 1);
|
||
var obj = JsonConvert.DeserializeObject<rcontact>(json);
|
||
var key = obj.type == 1 ? "rcontact1" : "rcontact0";
|
||
PushRcontact<rcontact>(obj, session.SessionID, key, obj.jobwxusername, obj.username);
|
||
}
|
||
else if (value.StartsWith("message:"))
|
||
{
|
||
var index = value.IndexOf(':');
|
||
var json = value.Substring(index + 1);
|
||
var obj = JsonConvert.DeserializeObject<message>(json);
|
||
PushMessage<message>(obj, session.SessionID, "message", obj.msgId);
|
||
}
|
||
}
|
||
|
||
private void _appServer_SessionClosed(WebSocketSession session, SuperSocket.SocketBase.CloseReason value)
|
||
{
|
||
var msg = string.Format("会话{3}关闭，关闭原因：{0} 来自：{1} 时间:{2:HH:MM:ss}", value, session.RemoteEndPoint, DateTime.Now, session.SessionID);
|
||
session.Logger.Info(msg);
|
||
session.Close();
|
||
if (_sessionList.ContainsKey(session.SessionID))
|
||
{
|
||
_sessionList.Remove(session.SessionID);
|
||
}
|
||
}
|
||
|
||
private void _appServer_NewSessionConnected(WebSocketSession session)
|
||
{
|
||
var msg = string.Format("新的会话连接 来自：{0} SessionID:{1} 时间:{2:HH:MM:ss}", session.RemoteEndPoint, session.SessionID, DateTime.Now);
|
||
_sessionList.Add(session.SessionID, 0);
|
||
session.Send("SessionID:" + session.SessionID);
|
||
session.Logger.Info(msg);
|
||
}
|
||
|
||
private void SendMsgToRemotePoint(string sessionId, string msg)
|
||
{
|
||
var allSession = _appServer.GetSessionByID(sessionId);
|
||
if (allSession != null)
|
||
allSession.TrySend(msg);
|
||
}
|
||
|
||
private async void PushRcontact<T>(T model, string sessionId, string key, string jobWxUserName, string userName)
|
||
{
|
||
await PushRcontactAsync(model, sessionId, key, jobWxUserName, userName);
|
||
}
|
||
private async Task PushRcontactAsync<T>(T model, string sessionId, string key, string jobWxUserName, string userName)
|
||
{
|
||
var isExists = await _redisSet.Contains(jobWxUserName, userName);
|
||
if (!isExists)
|
||
{
|
||
await _redisSet.AddAsync("wxid:" + jobWxUserName, userName); //写入微信好友关系
|
||
|
||
await PushListAndReplay<T>(model, sessionId, key, jobWxUserName, userName);
|
||
}
|
||
}
|
||
private async Task PushListAndReplay<T>(T model, string sessionId, string key, string jobWxUserName, string userName)
|
||
{
|
||
//写入队列
|
||
var result = await new RedisList<T>().LeftPushAsync(key, model);
|
||
if (result > 0)
|
||
{
|
||
//如果写入队列成功，回发确认信息:jobwxusername,username
|
||
SendMsgToRemotePoint(sessionId, "repRcon:" + jobWxUserName + "," + userName);
|
||
//发送订阅通知
|
||
await _sub.PublishAsync("sub:" + key, "");
|
||
}
|
||
}
|
||
|
||
private async void PushMessage<T>(T model, string sessionId, string key, int msgId)
|
||
{
|
||
await PushMessageAsync(model, sessionId, key, msgId);
|
||
}
|
||
|
||
private async Task PushMessageAsync<T>(T model, string sessionId, string key, int msgId)
|
||
{
|
||
var result = await new RedisList<T>().LeftPushAsync(key, model);
|
||
if (result > 0)
|
||
{
|
||
//如果写入队列成功，回发确认信息:msgSvrId
|
||
SendMsgToRemotePoint(sessionId, "repMsg:" + msgId);
|
||
//发送订阅通知
|
||
await _sub.PublishAsync("sub:" + key, "");
|
||
}
|
||
|
||
}
|
||
|
||
protected override void OnStart(string[] args)
|
||
{
|
||
#region socket服务端
|
||
//var serverConfig = new ServerConfig
|
||
//{
|
||
// Ip = Utility.GetSettingByKey("ServerIP"),
|
||
// Port = int.Parse(Utility.GetSettingByKey("ServerPort")),
|
||
// LogCommand = true,
|
||
// LogBasicSessionActivity = true,
|
||
// LogAllSocketException = true,
|
||
// MaxRequestLength = int.MaxValue
|
||
//};
|
||
//if (!_appServer.Setup(serverConfig))
|
||
//{
|
||
// LogHelper.Info("开启服务器失败");
|
||
// return;
|
||
//}
|
||
|
||
//if (!_appServer.Start())
|
||
//{
|
||
// LogHelper.Info("开启服务器失败");
|
||
// return;
|
||
//}
|
||
#endregion
|
||
|
||
//LogHelper.Info("Socket启动成功");
|
||
new Subscribe().Start();//启动订阅
|
||
LogHelper.Info("消息订阅启动");
|
||
_smsSend = new Thread(new ThreadStart(GetScheduleJob));
|
||
_smsSend.Start();
|
||
LogHelper.Info("服务已启动！");
|
||
}
|
||
Thread _smsSend;
|
||
protected override void OnStop()
|
||
{
|
||
try
|
||
{
|
||
if (_smsSend.ThreadState == System.Threading.ThreadState.Running)
|
||
_smsSend.Abort();
|
||
WX.CRM.Common.LogHelper.Info("服务停止！");
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
WX.CRM.Common.LogHelper.Error("服务停止异常" + ex.Message + ex.StackTrace);
|
||
}
|
||
//_appServer.Stop();
|
||
}
|
||
public void GetScheduleJob()
|
||
{
|
||
try
|
||
{
|
||
var processor = new JobSchedulingDataProcessor(true, true);
|
||
ISchedulerFactory sf = new Quartz.Impl.StdSchedulerFactory();
|
||
IScheduler sched = sf.GetScheduler();
|
||
Stream s = new StreamReader(string.Concat(AppDomain.CurrentDomain.BaseDirectory, "XML/JobConfig.xml"), System.Text.Encoding.GetEncoding("UTF-8")).BaseStream;
|
||
processor.ProcessStream(s, null);
|
||
processor.ScheduleJobs(new Hashtable(), sched, false);
|
||
sched.Start();
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
WX.CRM.Common.LogHelper.Error(string.Concat("WeChatServerServices.GetScheduleJob:", ex.Message, ex.StackTrace));
|
||
}
|
||
}
|
||
}
|
||
}
|