<commit_before>using System;
using System.Linq;
using System.Collections.Generic;


namespace Phoenix {

	public sealed class Socket {

		#region nested types

		public enum Events {
			Open,
			Close,
			Error,
			Message
		}

		public enum State {
			Closed,
			Closing,
			Connecting,
			Open,
		}

		public sealed class Options {
			// The default timeout to trigger push timeouts.
			public TimeSpan timeout = TimeSpan.FromSeconds(10);
			// The interval for rejoining an errored channel. Null means none
			public TimeSpan? channelRejoinInterval = TimeSpan.FromSeconds(1);
			// The interval to send a heartbeat message.
			public TimeSpan heartbeatInterval = TimeSpan.FromSeconds(30);
			// The optional function for specialized logging
			public ILogger logger = null;
			// The object responsible for performing delayed executions
			public IDelayedExecutor delayedExecutor = new TimerBasedExecutor();
		}

		#endregion


		#region events

		public delegate void OnOpenDelegate();
		public OnOpenDelegate OnOpen;

		public delegate void OnMessageDelegate(string message);
		public OnMessageDelegate OnMessage;

		public delegate void OnClosedDelegate(ushort code, string message);
		public OnClosedDelegate OnClose;

		public delegate void OnErrorDelegate(string message);
		public OnErrorDelegate OnError;

		#endregion


		#region properties

		public IWebsocket websocket { get; private set; }
		private readonly IWebsocketFactory websocketFactory;
		internal readonly Options opts;

		private uint? heartbeatTimer = null;
		private Dictionary<string, Channel> channels = new Dictionary<string, Channel>();

		public State state { get; private set; }

		#endregion


		public Socket(IWebsocketFactory factory, Options options = null) {

			websocketFactory = factory;
			opts = options ?? new Options();
		}


		#region private & internal methods

		private Uri BuildEndpointURL(string url, Dictionary<string, string> parameters) {
			// very primitive query string builder
			var stringParams = (parameters ?? new Dictionary<string, string>())
				.Select(pair => string.Format("{0}={1}", pair.Key, pair.Value))
				.ToArray();

			var query = string.Join("&", stringParams);

			var builder = new UriBuilder(string.Format("{0}/websocket", url));
			builder.Query = query;

			return builder.Uri;
		}

		private void SendHeartbeat() {

			if (state != State.Open) {
				return;
			}

			Push(new Message("phoenix", "heartbeat", null, null));
			heartbeatTimer = opts.delayedExecutor.Execute(SendHeartbeat, opts.heartbeatInterval);
		}

		private void RejoinChannels() {
			foreach (var channel in channels.Values) {
				channel.Rejoin();
			}
		}

		private void TriggerChannelError(string reason) {
			foreach (var channel in channels.Values) {
				channel.SocketTerminated(reason);
			}
		}

		private void CancelHeartbeat() {
			if (heartbeatTimer.HasValue) {
				opts.delayedExecutor.Cancel(heartbeatTimer.Value);
				heartbeatTimer = null;
			}
		}

		internal void Remove(Channel channel) {
			channels.Remove(channel.topic);
		}

		internal bool Push(Message msg) {

			var json = msg.Serialize();
			Log(LogLevel.Trace, "push", json);

			if (state == State.Open) {
				websocket.Send(json);
				return true;
			}

			return false;
		}

		// Logs the message. Override `this.logger` for specialized logging. noops by default
		internal void Log(LogLevel level, string kind, string msg) {
			if (opts.logger != null) {
				opts.logger.Log(level, kind, msg);
			}
		}

		#endregion


		#region public methods

		public void Disconnect(ushort? code = null, string reason = null) {

			CancelHeartbeat();
			TriggerChannelError("socket disconnect");

			if (websocket == null) {
				return;
			}

			// disables callbacks
			state = State.Closed;

			websocket.Close(code, reason);
			websocket = null;
		}

		// params - The params to send when connecting, for example `{user_id: userToken}`
		public void Connect(string url, Dictionary<string, string> parameters = null) {

			if (websocket != null) {
				Disconnect();
			}

			var config = new WebsocketConfiguration() {
				uri = BuildEndpointURL(url, parameters),
				onOpenCallback = WebsocketOnOpen,
				onCloseCallback = WebsocketOnClose,
				onErrorCallback = WebsocketOnError,
				onMessageCallback = WebsocketOnMessage
			};<commit_msg>allow heartbeat to be disabled. Avoid triggering channel error if socket is already closed<commit_after>using System;
using System.Linq;
using System.Collections.Generic;


namespace Phoenix {

	public sealed class Socket {

		#region nested types

		public enum Events {
			Open,
			Close,
			Error,
			Message
		}

		public enum State {
			Closed,
			Closing,
			Connecting,
			Open,
		}

		public sealed class Options {
			// The default timeout to trigger push timeouts.
			public TimeSpan timeout = TimeSpan.FromSeconds(10);
			// The interval for rejoining an errored channel. Null means none
			public TimeSpan? channelRejoinInterval = TimeSpan.FromSeconds(1);
			// The interval to send a heartbeat message. Null means disable
			public TimeSpan? heartbeatInterval = TimeSpan.FromSeconds(30);
			// The optional function for specialized logging
			public ILogger logger = null;
			// The object responsible for performing delayed executions
			public IDelayedExecutor delayedExecutor = new TimerBasedExecutor();
		}

		#endregion


		#region events

		public delegate void OnOpenDelegate();
		public OnOpenDelegate OnOpen;

		public delegate void OnMessageDelegate(string message);
		public OnMessageDelegate OnMessage;

		public delegate void OnClosedDelegate(ushort code, string message);
		public OnClosedDelegate OnClose;

		public delegate void OnErrorDelegate(string message);
		public OnErrorDelegate OnError;

		#endregion


		#region properties

		public IWebsocket websocket { get; private set; }
		private readonly IWebsocketFactory websocketFactory;
		internal readonly Options opts;

		private uint? heartbeatTimer = null;
		private Dictionary<string, Channel> channels = new Dictionary<string, Channel>();

		public State state { get; private set; }

		#endregion


		public Socket(IWebsocketFactory factory, Options options = null) {

			websocketFactory = factory;
			opts = options ?? new Options();
		}


		#region private & internal methods

		private Uri BuildEndpointURL(string url, Dictionary<string, string> parameters) {
			// very primitive query string builder
			var stringParams = (parameters ?? new Dictionary<string, string>())
				.Select(pair => string.Format("{0}={1}", pair.Key, pair.Value))
				.ToArray();

			var query = string.Join("&", stringParams);

			var builder = new UriBuilder(string.Format("{0}/websocket", url));
			builder.Query = query;

			return builder.Uri;
		}

		private void SendHeartbeat() {

			if (state != State.Open || opts.heartbeatInterval == null) {
				return;
			}

			Push(new Message("phoenix", "heartbeat", null, null));
			heartbeatTimer = opts.delayedExecutor.Execute(SendHeartbeat, opts.heartbeatInterval.Value);
		}

		private void RejoinChannels() {
			foreach (var channel in channels.Values) {
				channel.Rejoin();
			}
		}

		private void TriggerChannelError(string reason) {
			foreach (var channel in channels.Values) {
				channel.SocketTerminated(reason);
			}
		}

		private void CancelHeartbeat() {
			if (heartbeatTimer.HasValue) {
				opts.delayedExecutor.Cancel(heartbeatTimer.Value);
				heartbeatTimer = null;
			}
		}

		internal void Remove(Channel channel) {
			channels.Remove(channel.topic);
		}

		internal bool Push(Message msg) {

			var json = msg.Serialize();
			Log(LogLevel.Trace, "push", json);

			if (state == State.Open) {
				websocket.Send(json);
				return true;
			}

			return false;
		}

		// Logs the message. Override `this.logger` for specialized logging. noops by default
		internal void Log(LogLevel level, string kind, string msg) {
			if (opts.logger != null) {
				opts.logger.Log(level, kind, msg);
			}
		}

		#endregion


		#region public methods

		public void Disconnect(ushort? code = null, string reason = null) {

			if (websocket == null) {
				return;
			}

			CancelHeartbeat();
			TriggerChannelError("socket disconnect");

			// disables callbacks
			state = State.Closed;

			websocket.Close(code, reason);
			websocket = null;
		}

		// params - The params to send when connecting, for example `{user_id: userToken}`
		public void Connect(string url, Dictionary<string, string> parameters = null) {

			if (websocket != null) {
				Disconnect();
			}

			var config = new WebsocketConfiguration() {
				uri = BuildEndpointURL(url, parameters),
				onOpenCallback = WebsocketOnOpen,
				onCloseCallback = WebsocketOnClose,
				onErrorCallback = WebsocketOnError,
				onMessageCallback = WebsocketOnMessage
			};
{'fraction_non_alphanumeric': 0.09512946275807004, 'fraction_numerical': 0.001825025664423406, 'mean_word_length': 3.6837606837606836, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, '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': '759551', 'n_tokens_mistral': 3001, 'n_tokens_neox': 2442, 'n_words': 756}