using System;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net.WebSockets;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Websocket.Client.Exceptions;
using Websocket.Client.Logging;
using Websocket.Client.Models;
using Websocket.Client.Threading;
using Websocket.Client.Validations;

namespace Websocket.Client
{
	public class WebsocketClient : IWebsocketClient, IDisposable
	{
		public WebsocketClient(Uri url, Func<ClientWebSocket> clientFactory = null)
			: this(url, WebsocketClient.GetClientFactory(clientFactory))
		{
		}

		public WebsocketClient(Uri url, Func<Uri, CancellationToken, Task<WebSocket>> connectionFactory)
		{
			Validations.ValidateInput<Uri>(url, "url");
			this._url = url;
			Func<Uri, CancellationToken, Task<WebSocket>> func = connectionFactory;
			if (connectionFactory == null && (func = WebsocketClient.<>c.<>9__17_0) == null)
			{
				func = (WebsocketClient.<>c.<>9__17_0 = async delegate(Uri uri, CancellationToken token)
				{
					object client = new ClientWebSocket();
					await client.ConnectAsync(uri, token).ConfigureAwait(false);
					return client;
				});
			}
			this._connectionFactory = func;
		}

		public Uri Url
		{
			get
			{
				return this._url;
			}
			set
			{
				Validations.ValidateInput<Uri>(value, "Url");
				this._url = value;
			}
		}

		public IObservable<ResponseMessage> MessageReceived
		{
			get
			{
				return this._messageReceivedSubject.AsObservable<ResponseMessage>();
			}
		}

		public IObservable<ReconnectionInfo> ReconnectionHappened
		{
			get
			{
				return this._reconnectionSubject.AsObservable<ReconnectionInfo>();
			}
		}

		public IObservable<DisconnectionInfo> DisconnectionHappened
		{
			get
			{
				return this._disconnectedSubject.AsObservable<DisconnectionInfo>();
			}
		}

		public TimeSpan? ReconnectTimeout { get; set; } = new TimeSpan?(TimeSpan.FromMinutes(1.0));

		public TimeSpan? ErrorReconnectTimeout { get; set; } = new TimeSpan?(TimeSpan.FromMinutes(1.0));

		public bool IsReconnectionEnabled
		{
			get
			{
				return this._isReconnectionEnabled;
			}
			set
			{
				this._isReconnectionEnabled = value;
				if (this.IsStarted)
				{
					if (this._isReconnectionEnabled)
					{
						this.ActivateLastChance();
						return;
					}
					this.DeactivateLastChance();
				}
			}
		}

		public string Name { get; set; }

		public bool IsStarted { get; private set; }

		public bool IsRunning { get; private set; }

		public bool IsTextMessageConversionEnabled { get; set; } = true;

		public Encoding MessageEncoding { get; set; }

		public ClientWebSocket NativeClient
		{
			get
			{
				return this.GetSpecificOrThrow(this._client);
			}
		}

		public void Dispose()
		{
			this._disposing = true;
			WebsocketClient.Logger.Debug(this.L("Disposing.."));
			try
			{
				Channel<string> messagesTextToSendQueue = this._messagesTextToSendQueue;
				if (messagesTextToSendQueue != null)
				{
					messagesTextToSendQueue.Writer.Complete(null);
				}
				Channel<ArraySegment<byte>> messagesBinaryToSendQueue = this._messagesBinaryToSendQueue;
				if (messagesBinaryToSendQueue != null)
				{
					messagesBinaryToSendQueue.Writer.Complete(null);
				}
				Timer lastChanceTimer = this._lastChanceTimer;
				if (lastChanceTimer != null)
				{
					lastChanceTimer.Dispose();
				}
				CancellationTokenSource cancellation = this._cancellation;
				if (cancellation != null)
				{
					cancellation.Cancel();
				}
				CancellationTokenSource cancellationTotal = this._cancellationTotal;
				if (cancellationTotal != null)
				{
					cancellationTotal.Cancel();
				}
				WebSocket client = this._client;
				if (client != null)
				{
					client.Abort();
				}
				WebSocket client2 = this._client;
				if (client2 != null)
				{
					client2.Dispose();
				}
				CancellationTokenSource cancellation2 = this._cancellation;
				if (cancellation2 != null)
				{
					cancellation2.Dispose();
				}
				CancellationTokenSource cancellationTotal2 = this._cancellationTotal;
				if (cancellationTotal2 != null)
				{
					cancellationTotal2.Dispose();
				}
				this._messageReceivedSubject.OnCompleted();
				this._reconnectionSubject.OnCompleted();
			}
			catch (Exception ex)
			{
				WebsocketClient.Logger.Error(ex, this.L("Failed to dispose client, error: " + ex.Message), Array.Empty<object>());
			}
			this.IsRunning = false;
			this.IsStarted = false;
			this._disconnectedSubject.OnNext(DisconnectionInfo.Create(DisconnectionType.Exit, this._client, null));
			this._disconnectedSubject.OnCompleted();
		}

		public Task Start()
		{
			return this.StartInternal(false);
		}

		public Task StartOrFail()
		{
			return this.StartInternal(true);
		}

		public async Task<bool> Stop(WebSocketCloseStatus status, string statusDescription)
		{
			bool flag = await this.StopInternal(this._client, status, statusDescription, null, false, false).ConfigureAwait(false);
			this._disconnectedSubject.OnNext(DisconnectionInfo.Create(DisconnectionType.ByUser, this._client, null));
			return flag;
		}

		public async Task<bool> StopOrFail(WebSocketCloseStatus status, string statusDescription)
		{
			bool flag = await this.StopInternal(this._client, status, statusDescription, null, true, false).ConfigureAwait(false);
			this._disconnectedSubject.OnNext(DisconnectionInfo.Create(DisconnectionType.ByUser, this._client, null));
			return flag;
		}

		private static Func<Uri, CancellationToken, Task<WebSocket>> GetClientFactory(Func<ClientWebSocket> clientFactory)
		{
			WebsocketClient.<>c__DisplayClass65_0 CS$<>8__locals1 = new WebsocketClient.<>c__DisplayClass65_0();
			CS$<>8__locals1.clientFactory = clientFactory;
			if (CS$<>8__locals1.clientFactory == null)
			{
				return null;
			}
			return delegate(Uri uri, CancellationToken token)
			{
				WebsocketClient.<>c__DisplayClass65_0.<<GetClientFactory>b__0>d <<GetClientFactory>b__0>d;
				<<GetClientFactory>b__0>d.<>t__builder = AsyncTaskMethodBuilder<WebSocket>.Create();
				<<GetClientFactory>b__0>d.<>4__this = CS$<>8__locals1;
				<<GetClientFactory>b__0>d.uri = uri;
				<<GetClientFactory>b__0>d.token = token;
				<<GetClientFactory>b__0>d.<>1__state = -1;
				<<GetClientFactory>b__0>d.<>t__builder.Start<WebsocketClient.<>c__DisplayClass65_0.<<GetClientFactory>b__0>d>(ref <<GetClientFactory>b__0>d);
				return <<GetClientFactory>b__0>d.<>t__builder.Task;
			};
		}

		private async Task StartInternal(bool failFast)
		{
			if (this._disposing)
			{
				throw new WebsocketException(this.L("Client is already disposed, starting not possible"));
			}
			if (this.IsStarted)
			{
				WebsocketClient.Logger.Debug(this.L("Client already started, ignoring.."));
			}
			else
			{
				this.IsStarted = true;
				WebsocketClient.Logger.Debug(this.L("Starting.."));
				this._cancellation = new CancellationTokenSource();
				this._cancellationTotal = new CancellationTokenSource();
				await this.StartClient(this._url, this._cancellation.Token, ReconnectionType.Initial, failFast).ConfigureAwait(false);
				this.StartBackgroundThreadForSendingText();
				this.StartBackgroundThreadForSendingBinary();
			}
		}

		private async Task<bool> StopInternal(WebSocket client, WebSocketCloseStatus status, string statusDescription, CancellationToken? cancellation, bool failFast, bool byServer)
		{
			if (this._disposing)
			{
				throw new WebsocketException(this.L("Client is already disposed, stopping not possible"));
			}
			bool flag;
			if (!this.IsRunning)
			{
				WebsocketClient.Logger.Info(this.L("Client is already stopped"));
				flag = false;
			}
			else
			{
				bool result = false;
				if (client == null)
				{
					this.IsStarted = false;
					this.IsRunning = false;
					flag = false;
				}
				else
				{
					this.DeactivateLastChance();
					try
					{
						CancellationToken cancellationToken = cancellation ?? CancellationToken.None;
						this._stopping = true;
						if (byServer)
						{
							TaskAwaiter taskAwaiter = client.CloseOutputAsync(status, statusDescription, cancellationToken).GetAwaiter();
							if (!taskAwaiter.IsCompleted)
							{
								await taskAwaiter;
								TaskAwaiter taskAwaiter2;
								taskAwaiter = taskAwaiter2;
								taskAwaiter2 = default(TaskAwaiter);
							}
							taskAwaiter.GetResult();
						}
						else
						{
							await client.CloseAsync(status, statusDescription, cancellationToken);
						}
						result = true;
					}
					catch (Exception ex)
					{
						WebsocketClient.Logger.Error(ex, this.L("Error while stopping client, message: '" + ex.Message + "'"), Array.Empty<object>());
						if (failFast)
						{
							throw new WebsocketException("Failed to stop Websocket client, error: '" + ex.Message + "'", ex);
						}
					}
					finally
					{
						this.IsStarted = false;
						this.IsRunning = false;
						this._stopping = false;
					}
					flag = result;
				}
			}
			return flag;
		}

		private async Task StartClient(Uri uri, CancellationToken token, ReconnectionType type, bool failFast)
		{
			this.DeactivateLastChance();
			int num = 0;
			try
			{
				WebSocket webSocket = await this._connectionFactory(uri, token).ConfigureAwait(false);
				this._client = webSocket;
				this.Listen(this._client, token);
				this.IsRunning = true;
				this.IsStarted = true;
				this._reconnectionSubject.OnNext(ReconnectionInfo.Create(type));
				this._lastReceivedMsg = DateTime.UtcNow;
				this.ActivateLastChance();
			}
			catch (Exception obj)
			{
				num = 1;
			}
			object obj;
			if (num == 1)
			{
				Exception e = (Exception)obj;
				DisconnectionInfo disconnectionInfo = DisconnectionInfo.Create(DisconnectionType.Error, this._client, e);
				this._disconnectedSubject.OnNext(disconnectionInfo);
				if (disconnectionInfo.CancelReconnection)
				{
					WebsocketClient.Logger.Error(e, this.L("Exception while connecting. Reconnecting canceled by user, exiting. Error: '" + e.Message + "'"), Array.Empty<object>());
					return;
				}
				if (failFast)
				{
					throw new WebsocketException("Failed to start Websocket client, error: '" + e.Message + "'", e);
				}
				if (this.ErrorReconnectTimeout == null)
				{
					WebsocketClient.Logger.Error(e, this.L("Exception while connecting. Reconnecting disabled, exiting. Error: '" + e.Message + "'"), Array.Empty<object>());
					return;
				}
				TimeSpan value = this.ErrorReconnectTimeout.Value;
				WebsocketClient.Logger.Error(e, this.L("Exception while connecting. " + string.Format("Waiting {0} sec before next reconnection try. Error: '{1}'", value.TotalSeconds, e.Message)), Array.Empty<object>());
				await Task.Delay(value, token).ConfigureAwait(false);
				await this.Reconnect(ReconnectionType.Error, false, e).ConfigureAwait(false);
				e = null;
			}
			obj = null;
		}

		private bool IsClientConnected()
		{
			return this._client.State == WebSocketState.Open;
		}

		private async Task Listen(WebSocket client, CancellationToken token)
		{
			Exception causedException = null;
			try
			{
				ArraySegment<byte> buffer = new ArraySegment<byte>(new byte[4096]);
				for (;;)
				{
					byte[] resultArrayWithTrailing = null;
					int resultArraySize = 0;
					bool isResultArrayCloned = false;
					MemoryStream ms = null;
					WebSocketReceiveResult result;
					for (;;)
					{
						TaskAwaiter<WebSocketReceiveResult> taskAwaiter = client.ReceiveAsync(buffer, token).GetAwaiter();
						if (!taskAwaiter.IsCompleted)
						{
							await taskAwaiter;
							TaskAwaiter<WebSocketReceiveResult> taskAwaiter2;
							taskAwaiter = taskAwaiter2;
							taskAwaiter2 = default(TaskAwaiter<WebSocketReceiveResult>);
						}
						result = taskAwaiter.GetResult();
						byte[] array = buffer.Array;
						int count = result.Count;
						if (resultArrayWithTrailing == null)
						{
							resultArraySize += count;
							resultArrayWithTrailing = array;
							isResultArrayCloned = false;
						}
						else if (array != null)
						{
							if (ms == null)
							{
								ms = new MemoryStream();
								ms.Write(resultArrayWithTrailing, 0, resultArraySize);
							}
							ms.Write(array, buffer.Offset, count);
						}
						if (result.EndOfMessage)
						{
							break;
						}
						if (!isResultArrayCloned)
						{
							byte[] array2 = resultArrayWithTrailing;
							resultArrayWithTrailing = ((array2 != null) ? array2.ToArray<byte>() : null);
							isResultArrayCloned = true;
						}
					}
					MemoryStream memoryStream = ms;
					if (memoryStream != null)
					{
						memoryStream.Seek(0L, SeekOrigin.Begin);
					}
					ResponseMessage responseMessage;
					if (result.MessageType == WebSocketMessageType.Text && this.IsTextMessageConversionEnabled)
					{
						responseMessage = ResponseMessage.TextMessage((ms != null) ? this.GetEncoding().GetString(ms.ToArray()) : ((resultArrayWithTrailing != null) ? this.GetEncoding().GetString(resultArrayWithTrailing, 0, resultArraySize) : null));
						goto IL_021E;
					}
					if (result.MessageType == WebSocketMessageType.Close)
					{
						WebsocketClient.Logger.Trace(this.L("Received close message"));
						if (!this.IsStarted || this._stopping)
						{
							goto IL_03B9;
						}
						DisconnectionInfo disconnectionInfo = DisconnectionInfo.Create(DisconnectionType.ByServer, client, null);
						this._disconnectedSubject.OnNext(disconnectionInfo);
						if (!disconnectionInfo.CancelClosing)
						{
							goto IL_0335;
						}
						if (this.IsReconnectionEnabled)
						{
							break;
						}
					}
					else
					{
						if (ms != null)
						{
							responseMessage = ResponseMessage.BinaryMessage(ms.ToArray());
							goto IL_021E;
						}
						Array.Resize<byte>(ref resultArrayWithTrailing, resultArraySize);
						responseMessage = ResponseMessage.BinaryMessage(resultArrayWithTrailing);
						goto IL_021E;
					}
					IL_026F:
					if (client.State == WebSocketState.Open && !token.IsCancellationRequested)
					{
						continue;
					}
					goto IL_03E2;
					IL_021E:
					MemoryStream memoryStream2 = ms;
					if (memoryStream2 != null)
					{
						memoryStream2.Dispose();
					}
					WebsocketClient.Logger.Trace(this.L(string.Format("Received:  {0}", responseMessage)));
					this._lastReceivedMsg = DateTime.UtcNow;
					this._messageReceivedSubject.OnNext(responseMessage);
					resultArrayWithTrailing = null;
					ms = null;
					goto IL_026F;
				}
				throw new OperationCanceledException("Websocket connection was closed by server");
				IL_0335:
				TaskAwaiter<bool> taskAwaiter3 = this.StopInternal(client, WebSocketCloseStatus.NormalClosure, "Closing", new CancellationToken?(token), false, true).GetAwaiter();
				if (!taskAwaiter3.IsCompleted)
				{
					await taskAwaiter3;
					TaskAwaiter<bool> taskAwaiter4;
					taskAwaiter3 = taskAwaiter4;
					taskAwaiter4 = default(TaskAwaiter<bool>);
				}
				taskAwaiter3.GetResult();
				if (this.IsReconnectionEnabled && !this.ShouldIgnoreReconnection(client))
				{
					this.ReconnectSynchronized(ReconnectionType.Lost, false, null);
				}
				IL_03B9:
				return;
				IL_03E2:
				buffer = default(ArraySegment<byte>);
			}
			catch (TaskCanceledException causedException)
			{
			}
			catch (OperationCanceledException causedException)
			{
			}
			catch (ObjectDisposedException causedException)
			{
			}
			catch (Exception ex)
			{
				WebsocketClient.Logger.Error(ex, this.L("Error while listening to websocket stream, error: '" + ex.Message + "'"), Array.Empty<object>());
				causedException = ex;
			}
			if (!this.ShouldIgnoreReconnection(client) && this.IsStarted)
			{
				this.ReconnectSynchronized(ReconnectionType.Lost, false, causedException);
			}
		}

		private bool ShouldIgnoreReconnection(WebSocket client)
		{
			return (this._disposing || this._reconnecting || this._stopping) | (client != this._client);
		}

		private Encoding GetEncoding()
		{
			if (this.MessageEncoding == null)
			{
				this.MessageEncoding = Encoding.UTF8;
			}
			return this.MessageEncoding;
		}

		private ClientWebSocket GetSpecificOrThrow(WebSocket client)
		{
			if (client == null)
			{
				return null;
			}
			ClientWebSocket clientWebSocket = client as ClientWebSocket;
			if (clientWebSocket == null)
			{
				throw new WebsocketException("Cannot cast 'WebSocket' client to 'ClientWebSocket', provide correct type via factory or don't use this property at all.");
			}
			return clientWebSocket;
		}

		private string L(string msg)
		{
			string text = this.Name ?? "CLIENT";
			return "[WEBSOCKET " + text + "] " + msg;
		}

		private static ILog GetLogger()
		{
			ILog log;
			try
			{
				log = LogProvider.GetCurrentClassLogger();
			}
			catch (Exception ex)
			{
				Trace.WriteLine("[WEBSOCKET] Failed to initialize logger, disabling.. " + string.Format("Error: {0}", ex));
				log = LogProvider.NoOpLogger.Instance;
			}
			return log;
		}

		private DisconnectionType TranslateTypeToDisconnection(ReconnectionType type)
		{
			return (DisconnectionType)type;
		}

		public Task Reconnect()
		{
			return this.ReconnectInternal(false);
		}

		public Task ReconnectOrFail()
		{
			return this.ReconnectInternal(true);
		}

		private async Task ReconnectInternal(bool failFast)
		{
			if (!this.IsStarted)
			{
				WebsocketClient.Logger.Debug(this.L("Client not started, ignoring reconnection.."));
			}
			else
			{
				try
				{
					await this.ReconnectSynchronized(ReconnectionType.ByUser, failFast, null).ConfigureAwait(false);
				}
				finally
				{
					this._reconnecting = false;
				}
			}
		}

		private async Task ReconnectSynchronized(ReconnectionType type, bool failFast, Exception causedException)
		{
			IDisposable disposable = await this._locker.LockAsync();
			using (disposable)
			{
				await this.Reconnect(type, failFast, causedException);
			}
			IDisposable disposable2 = null;
		}

		private async Task Reconnect(ReconnectionType type, bool failFast, Exception causedException)
		{
			this.IsRunning = false;
			if (!this._disposing)
			{
				this._reconnecting = true;
				DisconnectionInfo disconnectionInfo = DisconnectionInfo.Create(this.TranslateTypeToDisconnection(type), this._client, causedException);
				if (type != ReconnectionType.Error)
				{
					this._disconnectedSubject.OnNext(disconnectionInfo);
					if (disconnectionInfo.CancelReconnection)
					{
						WebsocketClient.Logger.Info(this.L("Reconnecting canceled by user, exiting."));
					}
				}
				this._cancellation.Cancel();
				try
				{
					WebSocket client = this._client;
					if (client != null)
					{
						client.Abort();
					}
				}
				catch (Exception ex)
				{
					WebsocketClient.Logger.Error(ex, this.L("Exception while aborting client. Error: '" + ex.Message + "'"), Array.Empty<object>());
				}
				WebSocket client2 = this._client;
				if (client2 != null)
				{
					client2.Dispose();
				}
				if (this.IsReconnectionEnabled && !disconnectionInfo.CancelReconnection)
				{
					WebsocketClient.Logger.Debug(this.L("Reconnecting..."));
					this._cancellation = new CancellationTokenSource();
					ConfiguredTaskAwaitable.ConfiguredTaskAwaiter configuredTaskAwaiter = this.StartClient(this._url, this._cancellation.Token, type, failFast).ConfigureAwait(false).GetAwaiter();
					if (!configuredTaskAwaiter.IsCompleted)
					{
						await configuredTaskAwaiter;
						ConfiguredTaskAwaitable.ConfiguredTaskAwaiter configuredTaskAwaiter2;
						configuredTaskAwaiter = configuredTaskAwaiter2;
						configuredTaskAwaiter2 = default(ConfiguredTaskAwaitable.ConfiguredTaskAwaiter);
					}
					configuredTaskAwaiter.GetResult();
					this._reconnecting = false;
				}
				else
				{
					this.IsStarted = false;
					this._reconnecting = false;
				}
			}
		}

		private void ActivateLastChance()
		{
			this._lastChanceTimer = new Timer(new TimerCallback(this.LastChance), null, 1000, 1000);
		}

		private void DeactivateLastChance()
		{
			Timer lastChanceTimer = this._lastChanceTimer;
			if (lastChanceTimer != null)
			{
				lastChanceTimer.Dispose();
			}
			this._lastChanceTimer = null;
		}

		private void LastChance(object state)
		{
			if (this.IsReconnectionEnabled && this.ReconnectTimeout != null)
			{
				double num = Math.Abs(this.ReconnectTimeout.Value.TotalMilliseconds);
				if (Math.Abs(DateTime.UtcNow.Subtract(this._lastReceivedMsg).TotalMilliseconds) > num)
				{
					WebsocketClient.Logger.Debug(this.L(string.Format("Last message received more than {0:F} ms ago. Hard restart..", num)));
					this.DeactivateLastChance();
					this.ReconnectSynchronized(ReconnectionType.NoMessageReceived, false, null);
				}
				return;
			}
			this.DeactivateLastChance();
		}

		public void Send(string message)
		{
			Validations.ValidateInput(message, "message");
			this._messagesTextToSendQueue.Writer.TryWrite(message);
		}

		public void Send(byte[] message)
		{
			Validations.ValidateInput<byte[]>(message, "message");
			this._messagesBinaryToSendQueue.Writer.TryWrite(new ArraySegment<byte>(message));
		}

		public void Send(ArraySegment<byte> message)
		{
			Validations.ValidateInput<ArraySegment<byte>>(message, "message");
			this._messagesBinaryToSendQueue.Writer.TryWrite(message);
		}

		public Task SendInstant(string message)
		{
			Validations.ValidateInput(message, "message");
			return this.SendInternalSynchronized(message);
		}

		public Task SendInstant(byte[] message)
		{
			return this.SendInternalSynchronized(new ArraySegment<byte>(message));
		}

		public void StreamFakeMessage(ResponseMessage message)
		{
			Validations.ValidateInput<ResponseMessage>(message, "message");
			this._messageReceivedSubject.OnNext(message);
		}

		private async Task SendTextFromQueue()
		{
			try
			{
				for (;;)
				{
					ValueTaskAwaiter<bool> valueTaskAwaiter = this._messagesTextToSendQueue.Reader.WaitToReadAsync(default(CancellationToken)).GetAwaiter();
					if (!valueTaskAwaiter.IsCompleted)
					{
						await valueTaskAwaiter;
						ValueTaskAwaiter<bool> valueTaskAwaiter2;
						valueTaskAwaiter = valueTaskAwaiter2;
						valueTaskAwaiter2 = default(ValueTaskAwaiter<bool>);
					}
					if (!valueTaskAwaiter.GetResult())
					{
						break;
					}
					string message;
					while (this._messagesTextToSendQueue.Reader.TryRead(out message))
					{
						try
						{
							await this.SendInternalSynchronized(message).ConfigureAwait(false);
							continue;
						}
						catch (Exception ex)
						{
							WebsocketClient.Logger.Error(ex, this.L("Failed to send text message: '" + message + "'. Error: " + ex.Message), Array.Empty<object>());
							continue;
						}
						break;
					}
				}
			}
			catch (TaskCanceledException)
			{
			}
			catch (OperationCanceledException)
			{
			}
			catch (Exception ex2)
			{
				if (!this._cancellationTotal.IsCancellationRequested && !this._disposing)
				{
					WebsocketClient.Logger.Trace(this.L("Sending text thread failed, error: " + ex2.Message + ". Creating a new sending thread."));
					this.StartBackgroundThreadForSendingText();
				}
			}
		}

		private async Task SendBinaryFromQueue()
		{
			try
			{
				for (;;)
				{
					ValueTaskAwaiter<bool> valueTaskAwaiter = this._messagesBinaryToSendQueue.Reader.WaitToReadAsync(default(CancellationToken)).GetAwaiter();
					if (!valueTaskAwaiter.IsCompleted)
					{
						await valueTaskAwaiter;
						ValueTaskAwaiter<bool> valueTaskAwaiter2;
						valueTaskAwaiter = valueTaskAwaiter2;
						valueTaskAwaiter2 = default(ValueTaskAwaiter<bool>);
					}
					if (!valueTaskAwaiter.GetResult())
					{
						break;
					}
					ArraySegment<byte> message;
					while (this._messagesBinaryToSendQueue.Reader.TryRead(out message))
					{
						try
						{
							await this.SendInternalSynchronized(message).ConfigureAwait(false);
							continue;
						}
						catch (Exception ex)
						{
							WebsocketClient.Logger.Error(ex, this.L(string.Format("Failed to send binary message: '{0}'. Error: {1}", message, ex.Message)), Array.Empty<object>());
							continue;
						}
						break;
					}
				}
			}
			catch (TaskCanceledException)
			{
			}
			catch (OperationCanceledException)
			{
			}
			catch (Exception ex2)
			{
				if (!this._cancellationTotal.IsCancellationRequested && !this._disposing)
				{
					WebsocketClient.Logger.Trace(this.L("Sending binary thread failed, error: " + ex2.Message + ". Creating a new sending thread."));
					this.StartBackgroundThreadForSendingBinary();
				}
			}
		}

		private void StartBackgroundThreadForSendingText()
		{
			Task.Factory.StartNew<Task>((object _) => this.SendTextFromQueue(), TaskCreationOptions.LongRunning, this._cancellationTotal.Token);
		}

		private void StartBackgroundThreadForSendingBinary()
		{
			Task.Factory.StartNew<Task>((object _) => this.SendBinaryFromQueue(), TaskCreationOptions.LongRunning, this._cancellationTotal.Token);
		}

		private async Task SendInternalSynchronized(string message)
		{
			IDisposable disposable = await this._locker.LockAsync();
			using (disposable)
			{
				await this.SendInternal(message);
			}
			IDisposable disposable2 = null;
		}

		private async Task SendInternal(string message)
		{
			if (!this.IsClientConnected())
			{
				WebsocketClient.Logger.Debug(this.L("Client is not connected to server, cannot send:  " + message));
			}
			else
			{
				WebsocketClient.Logger.Trace(this.L("Sending:  " + message));
				byte[] bytes = this.GetEncoding().GetBytes(message);
				ArraySegment<byte> arraySegment = new ArraySegment<byte>(bytes);
				await this._client.SendAsync(arraySegment, WebSocketMessageType.Text, true, this._cancellation.Token).ConfigureAwait(false);
			}
		}

		private async Task SendInternalSynchronized(ArraySegment<byte> message)
		{
			IDisposable disposable = await this._locker.LockAsync();
			using (disposable)
			{
				await this.SendInternal(message);
			}
			IDisposable disposable2 = null;
		}

		private async Task SendInternal(ArraySegment<byte> message)
		{
			if (!this.IsClientConnected())
			{
				WebsocketClient.Logger.Debug(this.L(string.Format("Client is not connected to server, cannot send binary, length: {0}", message.Count)));
			}
			else
			{
				WebsocketClient.Logger.Trace(this.L(string.Format("Sending binary, length: {0}", message.Count)));
				await this._client.SendAsync(message, WebSocketMessageType.Binary, true, this._cancellation.Token).ConfigureAwait(false);
			}
		}

		private static readonly ILog Logger = WebsocketClient.GetLogger();

		private readonly WebsocketAsyncLock _locker = new WebsocketAsyncLock();

		private readonly Func<Uri, CancellationToken, Task<WebSocket>> _connectionFactory;

		private Uri _url;

		private Timer _lastChanceTimer;

		private DateTime _lastReceivedMsg = DateTime.UtcNow;

		private bool _disposing;

		private bool _reconnecting;

		private bool _stopping;

		private bool _isReconnectionEnabled = true;

		private WebSocket _client;

		private CancellationTokenSource _cancellation;

		private CancellationTokenSource _cancellationTotal;

		private readonly Subject<ResponseMessage> _messageReceivedSubject = new Subject<ResponseMessage>();

		private readonly Subject<ReconnectionInfo> _reconnectionSubject = new Subject<ReconnectionInfo>();

		private readonly Subject<DisconnectionInfo> _disconnectedSubject = new Subject<DisconnectionInfo>();

		private readonly Channel<string> _messagesTextToSendQueue = Channel.CreateUnbounded<string>(new UnboundedChannelOptions
		{
			SingleReader = true,
			SingleWriter = false
		});

		private readonly Channel<ArraySegment<byte>> _messagesBinaryToSendQueue = Channel.CreateUnbounded<ArraySegment<byte>>(new UnboundedChannelOptions
		{
			SingleReader = true,
			SingleWriter = false
		});
	}
}
