chenbin/tianrunCRM
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add

Browse Source
This commit is contained in:
陈彬
2020-07-04 14:41:25 +08:00
parent 70c346d2c1
commit a8f02e4da5
3748 changed files with 587372 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

506
Assets/trCRM/DistSpringWebsocketClient/Client.cs Normal file
View File

@@ -0,0 +1,506 @@
using System;
using System.Collections.Generic;
using System.Net.WebSockets;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using Coolape;
using XLua;
using UnityEngine;
using System.Threading.Tasks;
using System.Collections;
using System.IO;
namespace Dist.SpringWebsocket
{
//public delegate void Receive(StompFrame frame);
public class Client : MonoBehaviour
{
/// <summary>
/// 支持的stomp协议版本
/// </summary>
public static string StompPrototypeVersion = "accept-version:1.1,1.0";
/// <summary>
/// 心跳时间
/// </summary>
public static string HeartBeating = "heart-beat:10000,10000";
/// <summary>
/// 换行符，用于构造 Stomp 消息包
/// </summary>
public static Char LF = Convert.ToChar(10);
/// <summary>
/// 空字符，用于构造 Stomp 消息包
/// </summary>
public static Char NULL = Convert.ToChar(0);
/// <summary>
/// 当前连接类型
/// </summary>
public static string TYPE = "client";
public static MemoryStream receivedBuffer = new MemoryStream();
Queue<StompFrame> queueCallback = new Queue<StompFrame>();
private static string SubscriptionHeader = "subscription";
private static string DestinationHeader = "destination";
private static string ContentLengthHeader = "content-length";
private static string CID = "dist-connect";
private Dictionary<string, object> callbacks;
private object statusCallback;
private Dictionary<string, string> subscribes;
public ClientWebSocket socket;
private string url;
private static int COUNTER = 0;
public Boolean connected = false;
public Queue<ArraySegment<byte>> sendQueue = new Queue<ArraySegment<byte>>();
public static Client self;
public bool isSending = false;
public bool isDebug = false;
public Client()
{
self = this;
}
public async void init(string url, object callback)
{
//close();
this.url = url;
this.statusCallback = callback;
this.callbacks = new Dictionary<string, object>();
this.subscribes = new Dictionary<string, string>();
try
{
this.socket = new ClientWebSocket();
socket.ConnectAsync(new Uri(url), CancellationToken.None).Wait();
socket_Opened(this);
await StartReceiving(socket);
}
catch (Exception e)
{
Debug.LogError(e);
socket_Error(this, e.ToString());
}
}
public string URL
{
get { return url; }
set { this.url = value; }
}
async Task StartReceiving(ClientWebSocket client)
{
try
{
var array = new byte[1024 * 1024];
ArraySegment<byte> arraySegment = new ArraySegment<byte>(array);
var result = await client.ReceiveAsync(arraySegment, CancellationToken.None);
while (!result.CloseStatus.HasValue)
{
if (result.MessageType == WebSocketMessageType.Text)
{
//string msg = Encoding.UTF8.GetString(array, 0, result.Count);
//if (isDebug)
//{
// Debug.Log("receive:" + result.Count + "==\n" + msg);
//}
socket_MessageReceived(client, array, result.Count);
}
else if (result.MessageType == WebSocketMessageType.Close)
{
socket_Closed(this);
break;
}
if (client == null || client.State == WebSocketState.Aborted
|| client.State == WebSocketState.CloseSent
|| client.State == WebSocketState.CloseReceived
|| client.State == WebSocketState.Closed
)
{
socket_Error(this, "State=" + client.State);
break;
}
result = await client.ReceiveAsync(arraySegment, CancellationToken.None);
}
}
catch (Exception e)
{
Debug.LogError(e);
socket_Error(this, e.ToString());
}
}
async Task startSending(ClientWebSocket client)
{
try
{
if (client == null || client.State == WebSocketState.Aborted
|| client.State == WebSocketState.CloseSent
|| client.State == WebSocketState.CloseReceived
|| client.State == WebSocketState.Closed)
{
Debug.LogWarning("startSending=State=" + ((client != null) ? client.State.ToString() : ""));
return;
}
isSending = true;
while (sendQueue.Count > 0)
{
if (client == null || client.State == WebSocketState.Aborted
|| client.State == WebSocketState.CloseSent
|| client.State == WebSocketState.CloseReceived
|| client.State == WebSocketState.Closed)
{
socket_Error(this, "State=" + client.State);
break;
}
if (sendQueue.Count > 0)
{
var array = sendQueue.Dequeue();
await socket.SendAsync(array, WebSocketMessageType.Text, true, CancellationToken.None);
}
}
isSending = false;
}
catch (Exception e)
{
isSending = false;
Debug.LogError(e);
socket_Error(this, e.ToString());
}
}
async void _send(string content)
{
if (isDebug)
{
Debug.Log("send==\n" + content);
}
ArraySegment<byte> array = new ArraySegment<byte>(Encoding.UTF8.GetBytes(content));
sendQueue.Enqueue(array);
if (!isSending)
{
await startSending(this.socket);
}
}
public void Connect(Dictionary<string, string> headers, object callback)
{
if (!this.connected)
{
this.callbacks[CID] = callback;
string data = StompCommandEnum.CONNECT.ToString() + LF;
if (headers != null)
{
foreach (string key in headers.Keys)
{
data += key + ":" + headers[key] + LF;
}
}
data += StompPrototypeVersion + LF
+ HeartBeating + LF + LF + NULL;
_send(data);
}
}
public void Send(string destination, string content)
{
this.Send(destination, new Dictionary<string, string>(), content);
}
public void Send(string destination, Dictionary<string, string> headers, string content)
{
string data = StompCommandEnum.SEND.ToString() + LF;
if (headers != null)
{
foreach (string key in headers.Keys)
{
data += key + ":" + headers[key] + LF;
}
}
data += DestinationHeader + ":" + destination + LF
+ ContentLengthHeader + ":" + GetByteCount(content) + LF + LF
+ content + NULL;
_send(data);
}
public void Send(string destination, LuaTable headers, string content)
{
string data = StompCommandEnum.SEND.ToString() + LF;
if (headers != null)
{
headers.ForEach<string, object>((key, val) =>
{
data += key + ":" + val.ToString() + LF;
});
}
data += DestinationHeader + ":" + destination + LF
+ ContentLengthHeader + ":" + GetByteCount(content) + LF + LF
+ content + NULL;
_send(data);
}
public void Subscribe(string destination, object callback)
{
this.Subscribe(destination, new Dictionary<string, string>(), callback);
}
public void Subscribe(string destination, Dictionary<string, string> headers, object callback)
{
lock (this)
{
if (!this.subscribes.ContainsKey(destination))
{
string id = "sub-" + COUNTER++;
this.callbacks.Add(id, callback);
this.subscribes.Add(destination, id);
string data = StompCommandEnum.SUBSCRIBE.ToString() + LF + "id:" + id + LF;
foreach (string key in headers.Keys)
{
data += key + ":" + headers[key] + LF;
}
data += DestinationHeader + ":" + destination + LF + LF + NULL;
_send(data);
}
}
}
public void UnSubscribe(string destination)
{
if (this.subscribes.ContainsKey(destination))
{
_send(StompCommandEnum.UNSUBSCRIBE.ToString() + LF + "id:" + this.subscribes[destination] + LF + LF + NULL);
this.callbacks.Remove(this.subscribes[destination]);
this.subscribes.Remove(destination);
}
}
public void DisConnect()
{
_send(StompCommandEnum.DISCONNECT.ToString() + LF + LF + NULL);
ArrayList list = new ArrayList();
list.AddRange(subscribes.Keys);
foreach (var key in list)
{
UnSubscribe(key.ToString());
}
this.callbacks.Clear();
this.subscribes.Clear();
this.connected = false;
isSending = false;
}
public bool IsSubscribed(string destination)
{
return this.subscribes.ContainsKey(destination);
}
private int GetByteCount(string content)
{
return Regex.Split(Uri.EscapeUriString(content), "%..|.").Length - 1;
}
private StompFrame TransformResultFrame(string content)
{
lock (this)
{
StompFrame frame = new StompFrame();
//string[] matches = Regex.Split(content, "" + NULL + LF + "*");
//foreach (var line in matches)
//{
//if (line.Length > 0)
//{
this.HandleSingleLine(content, frame);
// }
//}
return frame;
}
}
private void HandleSingleLine(string line, StompFrame frame)
{
int divider = line.IndexOf("" + LF + LF);
if (divider >= 0)
{
string[] headerLines = Regex.Split(line.Substring(0, divider), "" + LF);
frame.Code = (StatusCodeEnum)Enum.Parse(typeof(StatusCodeEnum), headerLines[0]);
for (int i = 1; i < headerLines.Length; i++)
{
int index = headerLines[i].IndexOf(":");
string key = headerLines[i].Substring(0, index);
string value = headerLines[i].Substring(index + 1);
frame.AddHeader(Regex.Replace(key, @"^\s+|\s+$", ""), Regex.Replace(value, @"^\s+|\s+$", ""));
}
frame.Content = line.Substring(divider + 2);
}
}
private void socket_Error(object sender, string errMsg)
{
if (isDebug)
{
Debug.LogWarning("socket_Error==" + errMsg);
}
//close();
//this.connected = false;
queueCallback.Enqueue(new StompFrame(StatusCodeEnum.SERVERERROR, errMsg, null, statusCallback));
}
void socket_Closed(object sender)
{
if (isDebug)
{
Debug.LogWarning("socket_Closed");
}
this.connected = false;
if (socket != null)
{
socket.Abort();
socket.Dispose();
socket = null;
}
//this.socket.Dispose();
queueCallback.Enqueue(new StompFrame(StatusCodeEnum.SERVERCLOSED, "与服务器断开连接", null, statusCallback));
}
private void socket_Opened(object sender)
{
queueCallback.Enqueue(new StompFrame(StatusCodeEnum.OPENSERVER, "成功连接到服务器", null, statusCallback));
}
public static int __maxLen = 1024 * 1024;
byte[] tmpBuffer = new byte[__maxLen];
StringBuilder inputSb = new StringBuilder();
void socket_MessageReceived(object sender, byte[] bytes, int len)
{
receivedBuffer.Write(bytes, 0, len);
int totalLen = (int)(receivedBuffer.Length);
receivedBuffer.Position = 0;
receivedBuffer.Read(tmpBuffer, 0, totalLen);
receivedBuffer.Position = totalLen;
string msg = Encoding.UTF8.GetString(tmpBuffer, 0, totalLen);
inputSb.Append(msg);
while (true)
{
int index = inputSb.ToString().IndexOf(NULL);
if (index < 0)
{
//  说明数据包还不完整
break;
}
string content = inputSb.ToString().Substring(0, index);
inputSb.Remove(0, index + 1);
if(isDebug)
{
Debug.LogWarning(content);
}
StompFrame frame = this.TransformResultFrame(content);
object callback;
switch (frame.Code)
{
case StatusCodeEnum.CONNECTED:
connected = true;
callbacks.TryGetValue(CID, out callback);
frame.callback = callback;
queueCallback.Enqueue(frame);
break;
case StatusCodeEnum.MESSAGE:
callbacks.TryGetValue(frame.GetHeader(SubscriptionHeader), out callback);
frame.callback = callback;
queueCallback.Enqueue(frame);
break;
default:
Debug.LogError(frame.Code);
break;
}
}
// left msg proc
msg = inputSb.ToString();
inputSb.Clear();
int leftLen = 0;
if (!string.IsNullOrEmpty(msg))
{
byte[] leftBytes = Encoding.UTF8.GetBytes(msg);
leftLen = leftBytes.Length;
}
if (leftLen > 0)
{
receivedBuffer.Read(tmpBuffer, 0, leftLen);
receivedBuffer.Position = 0;
receivedBuffer.Write(tmpBuffer, 0, leftLen);
receivedBuffer.SetLength(leftLen);
} else
{
receivedBuffer.Position = 0;
receivedBuffer.SetLength(0);
}
}
//===============================================
StompFrame _stompframe;
private void Update()
{
if (queueCallback.Count > 0)
{
_stompframe = queueCallback.Dequeue();
if (_stompframe != null)
{
Utl.doCallback(_stompframe.callback, _stompframe);
}
}
}
public void close()
{
try
{
receivedBuffer.Position = 0;
receivedBuffer.SetLength(0);
if (connected)
{
connected = false;
isSending = false;
if (isDebug)
{
Debug.Log("close===");
}
if (socket == null || socket.State == WebSocketState.Aborted
|| socket.State == WebSocketState.CloseSent
|| socket.State == WebSocketState.CloseReceived
|| socket.State == WebSocketState.Closed)
{
if (socket != null)
{
if (isDebug)
{
Debug.LogWarning("socket.Abort");
}
//socket.CloseOutputAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None).Wait();
//socket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None).Wait();
socket.Abort();
socket.Dispose();
socket = null;
}
}
else
{
if (isDebug)
{
Debug.LogWarning("send DisConnect");
}
DisConnect();
}
queueCallback.Clear();
}
}
catch (Exception e)
{
Debug.LogError(e);
}
}
private void OnDestroy()
{
close();
}
}
}

11
Assets/trCRM/DistSpringWebsocketClient/Client.cs.meta Normal file
View File

@@ -0,0 +1,11 @@
fileFormatVersion: 2
guid: ed0b65beeb03b4ee4a5a604188b24571
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant:

8
Assets/trCRM/DistSpringWebsocketClient/Properties.meta Normal file
View File

@@ -0,0 +1,8 @@
fileFormatVersion: 2
guid: 02773c15570e142f9a69b26adda93538
folderAsset: yes
DefaultImporter:
externalObjects: {}
userData:
assetBundleName:
assetBundleVariant:

36
Assets/trCRM/DistSpringWebsocketClient/Properties/AssemblyInfo.cs Normal file
View File

@@ -0,0 +1,36 @@
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
// 有关程序集的常规信息通过以下
// 特性集控制。更改这些特性值可修改
// 与程序集关联的信息。
[assembly: AssemblyTitle("DistSpringWebsocketClient")]
[assembly: AssemblyDescription("")]
[assembly: AssemblyConfiguration("")]
[assembly: AssemblyCompany("")]
[assembly: AssemblyProduct("DistSpringWebsocketClient")]
[assembly: AssemblyCopyright("Copyright © 2016")]
[assembly: AssemblyTrademark("")]
[assembly: AssemblyCulture("")]
// 将 ComVisible 设置为 false 使此程序集中的类型
// 对 COM 组件不可见。 如果需要从 COM 访问此程序集中的类型，
// 则将该类型上的 ComVisible 特性设置为 true。
[assembly: ComVisible(false)]
// 如果此项目向 COM 公开，则下列 GUID 用于类型库的 ID
[assembly: Guid("c786064a-8aae-4cb8-b686-c8d764cace9f")]
// 程序集的版本信息由下面四个值组成:
//
// 主版本
// 次版本
// 生成号
// 修订号
//
// 可以指定所有这些值，也可以使用“生成号”和“修订号”的默认值，
// 方法是按如下所示使用“*”:
// [assembly: AssemblyVersion("1.0.*")]
[assembly: AssemblyVersion("1.0.0.0")]
[assembly: AssemblyFileVersion("1.0.0.0")]

11
Assets/trCRM/DistSpringWebsocketClient/Properties/AssemblyInfo.cs.meta Normal file
View File

@@ -0,0 +1,11 @@
fileFormatVersion: 2
guid: d2971e4633a07404294ecdad29fa347b
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant:

18
Assets/trCRM/DistSpringWebsocketClient/StatusCodeEnum.cs Normal file
View File

@@ -0,0 +1,18 @@
using System;
using System.Collections.Generic;
using System.Text;
namespace Dist.SpringWebsocket
{
public enum StatusCodeEnum
{
OPENSERVER,
CANNOTOPENSERVER,
SERVERCLOSED,
SERVERERROR,
CONNECTED,
MESSAGE,
RECEIPT,
ERROR
}
}

11
Assets/trCRM/DistSpringWebsocketClient/StatusCodeEnum.cs.meta Normal file
View File

@@ -0,0 +1,11 @@
fileFormatVersion: 2
guid: 2610d7289bbc24db29764c10ae1e6dfd
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant:

24
Assets/trCRM/DistSpringWebsocketClient/StompCommandEnum.cs Normal file
View File

@@ -0,0 +1,24 @@
using System;
using System.Collections.Generic;
using System.Text;
namespace Dist.SpringWebsocket
{
public enum StompCommandEnum
{
CONNECT,
CONNECTED,
SEND,
SUBSCRIBE,
UNSUBSCRIBE,
ACK,
NACK,
BEGIN,
COMMIT,
ABORT,
DISCONNECT,
MESSAGE,
RECEIPT,
ERROR
}
}

11
Assets/trCRM/DistSpringWebsocketClient/StompCommandEnum.cs.meta Normal file
View File

@@ -0,0 +1,11 @@
fileFormatVersion: 2
guid: 267a89c17fbb243d7a4b8ecf5bd50651
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant:

67
Assets/trCRM/DistSpringWebsocketClient/StompFrame.cs Normal file
View File

@@ -0,0 +1,67 @@
//using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.Text;
namespace Dist.SpringWebsocket
{
public class StompFrame
{
private StatusCodeEnum code;
private string content;
private Dictionary<string,string> headers;
public object callback;
public StompFrame() {
}
public StompFrame(StatusCodeEnum code):this()
{
this.code = code;
}
public StompFrame(StatusCodeEnum code, string content) :this(code)
{
this.content = content;
}
public StompFrame(StatusCodeEnum code, string content, Dictionary<string, string> headers) : this(code, content)
{
this.headers = headers;
}
public StompFrame(StatusCodeEnum code, string content, Dictionary<string, string> headers, object callback) : this(code, content, headers)
{
this.callback = callback;
}
public void AddHeader(string key, string value)
{
if (this.headers == null)
{
this.headers = new Dictionary<string, string>();
}
this.headers.Add(key, value);
}
public string GetHeader(string key)
{
return this.headers[key];
}
public bool ContainsHeader(string key)
{
return this.headers.ContainsKey(key);
}
public StatusCodeEnum Code
{
get { return code; }
set { code = value; }
}
public string Content
{
get { return content; }
set { content = value; }
}
public Dictionary<string, string> Headers
{
get { return headers; }
set { headers = value; }
}
}
}

11
Assets/trCRM/DistSpringWebsocketClient/StompFrame.cs.meta Normal file
View File

@@ -0,0 +1,11 @@
fileFormatVersion: 2
guid: c83c2193c2d31458aa37954d2c528bde
MonoImporter:
externalObjects: {}
serializedVersion: 2
defaultReferences: []
executionOrder: 0
icon: {instanceID: 0}
userData:
assetBundleName:
assetBundleVariant: