<commit_before>﻿#if !DOTNET5_4
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using WS4NetWebSocket = WebSocket4Net.WebSocket;

namespace Discord.Net.WebSockets
{
    internal class WS4NetEngine : IWebSocketEngine
    {
        private readonly DiscordConfig _config;
        private readonly Logger _logger;
        private readonly ConcurrentQueue<string> _sendQueue;
        private readonly WebSocket _parent;
        private WS4NetWebSocket _webSocket;
        private ManualResetEventSlim _waitUntilConnect;

        public event EventHandler<WebSocketBinaryMessageEventArgs> BinaryMessage;
        public event EventHandler<WebSocketTextMessageEventArgs> TextMessage;
        private void RaiseBinaryMessage(byte[] data)
        {
            if (BinaryMessage != null)
                BinaryMessage(this, new WebSocketBinaryMessageEventArgs(data));
        }
        private void RaiseTextMessage(string msg)
        {
            if (TextMessage != null)
                TextMessage(this, new WebSocketTextMessageEventArgs(msg));
        }

        internal WS4NetEngine(WebSocket parent, DiscordConfig config, Logger logger)
        {
            _parent = parent;
            _config = config;
            _logger = logger;
            _sendQueue = new ConcurrentQueue<string>();
            _waitUntilConnect = new ManualResetEventSlim();
        }

        public Task Connect(string host, CancellationToken cancelToken)
        {
            _webSocket = new WS4NetWebSocket(host);
            _webSocket.EnableAutoSendPing = false;
            _webSocket.NoDelay = true;
            _webSocket.Proxy = null; //Disable

            _webSocket.DataReceived += (s, e) =>
            {
                RaiseBinaryMessage(e.Data);
            };
            _webSocket.MessageReceived += (s, e) =>
            {
                RaiseTextMessage(e.Message);
            };
            _webSocket.Error += async (s, e) =>
            {
                _logger.Log(LogSeverity.Error, "WebSocket Error", e.Exception);
                await _parent.SignalDisconnect(e.Exception, isUnexpected: true).ConfigureAwait(false);
                _waitUntilConnect.Set();
            };
            _webSocket.Closed += async (s, e) =>
            {
                /*string code = e.WasClean ? e.Code.ToString() : "Unexpected";
                string reason = e.Reason != "" ? e.Reason : "No Reason";*/
                var ex = new Exception($"Got Close Message");// ({code}): {reason}");
                await _parent.SignalDisconnect(ex, isUnexpected: false/*true*/).ConfigureAwait(false);
                _waitUntilConnect.Set();
            };
            _webSocket.Opened += (s, e) =>
            {
                _waitUntilConnect.Set();
            };

            _waitUntilConnect.Reset();
            _webSocket.Open();
            _waitUntilConnect.Wait(cancelToken);
            return TaskHelper.CompletedTask;
        }

        public Task Disconnect()
        {
            string ignored;
            while (_sendQueue.TryDequeue(out ignored)) { }

            var socket = _webSocket;
            _webSocket = null;
            if (socket != null)
                socket.Close();

            return TaskHelper.CompletedTask;
        }

        public IEnumerable<Task> GetTasks(CancellationToken cancelToken)
        {
            return new Task[]
            {
                SendAsync(cancelToken)
            };
        }

        private Task SendAsync(CancellationToken cancelToken)
        {
            var sendInterval = _config.WebSocketInterval;
            return Task.Run(async () =>
            {
                try
                {
                    while (!cancelToken.IsCancellationRequested)
                    {
                        string json;<commit_msg>Fixed race condition in websocket<commit_after>﻿#if !DOTNET5_4
using SuperSocket.ClientEngine;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using WebSocket4Net;
using WS4NetWebSocket = WebSocket4Net.WebSocket;

namespace Discord.Net.WebSockets
{
    internal class WS4NetEngine : IWebSocketEngine
    {
        private readonly DiscordConfig _config;
        private readonly Logger _logger;
        private readonly ConcurrentQueue<string> _sendQueue;
        private readonly WebSocket _parent;
        private WS4NetWebSocket _webSocket;
        private ManualResetEventSlim _waitUntilConnect;

        public event EventHandler<WebSocketBinaryMessageEventArgs> BinaryMessage;
        public event EventHandler<WebSocketTextMessageEventArgs> TextMessage;
        private void RaiseBinaryMessage(byte[] data)
        {
            if (BinaryMessage != null)
                BinaryMessage(this, new WebSocketBinaryMessageEventArgs(data));
        }
        private void RaiseTextMessage(string msg)
        {
            if (TextMessage != null)
                TextMessage(this, new WebSocketTextMessageEventArgs(msg));
        }

        internal WS4NetEngine(WebSocket parent, DiscordConfig config, Logger logger)
        {
            _parent = parent;
            _config = config;
            _logger = logger;
            _sendQueue = new ConcurrentQueue<string>();
            _waitUntilConnect = new ManualResetEventSlim();
        }

        public Task Connect(string host, CancellationToken cancelToken)
        {
            _webSocket = new WS4NetWebSocket(host);
            _webSocket.EnableAutoSendPing = false;
            _webSocket.NoDelay = true;
            _webSocket.Proxy = null; //Disable

            _webSocket.DataReceived += OnWebSocketBinary;
            _webSocket.MessageReceived += OnWebSocketText;
            _webSocket.Error += OnWebSocketError;
            _webSocket.Closed += OnWebSocketClosed;
            _webSocket.Opened += OnWebSocketOpened;

            _waitUntilConnect.Reset();
            _webSocket.Open();
            _waitUntilConnect.Wait(cancelToken);
            return TaskHelper.CompletedTask;
        }

        public Task Disconnect()
        {
            string ignored;
            while (_sendQueue.TryDequeue(out ignored)) { }

            var socket = _webSocket;
            _webSocket = null;
            if (socket != null)
            {
                //We dont want a slow disconnect to mess up the next connection, so lets just unregister events
                socket.DataReceived -= OnWebSocketBinary;
                socket.MessageReceived -= OnWebSocketText;
                socket.Error -= OnWebSocketError;
                socket.Closed -= OnWebSocketClosed;
                socket.Opened -= OnWebSocketOpened;
                socket.Close();
            }

            return TaskHelper.CompletedTask;
        }

        private async void OnWebSocketError(object sender, ErrorEventArgs e)
        {
            await _parent.SignalDisconnect(e.Exception, isUnexpected: true).ConfigureAwait(false);
            _waitUntilConnect.Set();
        }
        private async void OnWebSocketClosed(object sender, EventArgs e)
        {
            var ex = new Exception($"Connection lost or close message received.");
            await _parent.SignalDisconnect(ex, isUnexpected: false/*true*/).ConfigureAwait(false);
            _waitUntilConnect.Set();
        }
        private void OnWebSocketOpened(object sender, EventArgs e)
        {
            _waitUntilConnect.Set();
        }
        private void OnWebSocketText(object sender, MessageReceivedEventArgs e)
        {
            RaiseTextMessage(e.Message);
        }
        private void OnWebSocketBinary(object sender, DataReceivedEventArgs e)
        {
            RaiseBinaryMessage(e.Data);
        }

        public IEnumerable<Task> GetTasks(CancellationToken cancelToken)
        {
            return new Task[]
            {
                SendAsync(cancelToken)
            };
        }

        private Task SendAsync(CancellationToken cancelToken)
        {
            var sendInterval = _config.WebSocketInterval;
            return Task.Run(async () =>
            {
                try
                {
                    while (!cancelToken.IsCancellationRequested)
                    {
                        string json;
{'fraction_non_alphanumeric': 0.07741010631943615, 'fraction_numerical': 0.0020308206904790346, 'mean_word_length': 1.88093599449415, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3824972', 'n_tokens_mistral': 2092, 'n_tokens_neox': 2025, 'n_words': 455}