<commit_before>using System;
using EasyZMq.Configuration;
using EasyZMq.Logging;
using NetMQ;
using NetMQ.Monitoring;

namespace EasyZMq.Sockets
{
    public abstract class EasyZMqReceiverSocket : IEasyZMqReceiverSocket
    {
        private readonly EasyZMqConfiguration _configuration;
        private readonly ILogger _logger;
        private NetMQContext _context;
        private NetMQSocket _socket;
        private NetMQMonitor _monitor;
        private Poller _poller;

        public event Action Connected;
        public event Action Disconnected;
        public event Action ConnectRetried;

        protected EasyZMqReceiverSocket(EasyZMqConfiguration configuration, NetMQContext context, NetMQSocket socket)
        {
            _configuration = configuration;
            _logger = configuration.EasyZMqLoggerFactory.GetLogger(typeof (EasyZMqReceiverSocket));
            _context = context;
            _socket = socket;

            CreatePoller(socket);
            CreateMonitor(context, socket, _poller);
            ConfigureSocket(socket);
        }

        public Uri Uri
        {
            get { return _configuration.AddressBinder.Uri; }
        }

        public void Start()
        {
            _configuration.AddressBinder.ConnectOrBindAddress(_socket);

            _poller.PollTillCancelledNonBlocking();
        }

        public abstract void OnMessageReceived<T>(T message);

        private void CreatePoller(ISocketPollable socket)
        {
            _poller = new Poller(socket);
        }

        private void CreateMonitor(NetMQContext context, NetMQSocket socket, Poller poller)
        {
            _monitor = new NetMQMonitor(context, socket, string.Format("inproc://{0}.inproc", Guid.NewGuid()),
                SocketEvents.Connected | SocketEvents.Disconnected | SocketEvents.ConnectRetried);

            _monitor.Connected += Monitor_Connected;
            _monitor.Disconnected += Monitor_Disconnected;
            _monitor.ConnectRetried += Monitor_ConnectRetried;

            _monitor.AttachToPoller(poller);
        }

        private void ConfigureSocket(NetMQSocket socket)
        {
            socket.Options.TcpKeepalive = true;
            socket.Options.TcpKeepaliveIdle = TimeSpan.FromSeconds(5);
            socket.Options.TcpKeepaliveInterval = TimeSpan.FromSeconds(1);
            socket.ReceiveReady += Subscriber_ReceiveReady;
        }

        private void Subscriber_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            bool more;
            e.Socket.ReceiveFrameString(out more);

            if (!more)
            {
                _logger.Warning("Invalid message received! The message was discarded.");
                return;
            }

            var bytes = e.Socket.ReceiveFrameBytes();

            DeserializeAndDispatch(bytes);
        }

        private void Monitor_Connected(object sender, NetMQMonitorSocketEventArgs e)
        {
            var handler = Connected;
            if (handler != null)
            {
                handler();
            }
        }

        private void Monitor_Disconnected(object sender, NetMQMonitorSocketEventArgs e)
        {
            var handler = Disconnected;
            if (handler != null)
            {
                handler();
            }
        }

        private void Monitor_ConnectRetried(object sender, NetMQMonitorIntervalEventArgs e)
        {
            var handler = ConnectRetried;
            if (handler != null)
            {
                handler();
            }
        }

        private void DeserializeAndDispatch(byte[] bytes)
        {
            try
            {
                dynamic message = _configuration.Serializer.Deserialize(bytes);
                OnMessageReceived(message);
            }
            catch (Exception ex)
            {
                _logger.Error("Deserialization and dispatch of incoming message failed.", ex);
            }
        }

        private bool _disposedValue;
        protected virtual void Dispose(bool disposing)
        {
            if (!_disposedValue)
            {
                if (disposing)
                {
                    if (_poller != null)
                    {
                        _poller.CancelAndJoin();

                        DisposeMonitor();
                        DisposePoller();
                        DisposeSocket();
                        DisposeContext();
                    }
                }

                _disposedValue = true;
            }
        }

        public void Dispose()
        {
            Dispose(true);
        }

        private void DisposeMonitor()
        {
            _monitor.Connected -= Monitor_Connected;
            _monitor.Disconnected -= Monitor_Disconnected;
            _monitor.ConnectRetried -= Monitor_ConnectRetried;
            _monitor.Dispose();
            _monitor = null;
        }

        private void DisposePoller()
        {
            _poller.Dispose();
            _poller = null;
        }

        private void DisposeSocket()
        {
            _socket.ReceiveReady -= Subscriber_ReceiveReady;
            _socket.Dispose();
            _socket = null;
        }<commit_msg>Fixed implementation of IDisposable. Throw ObjectDisposedException if trying to use object after it has been disposed<commit_after>using System;
using EasyZMq.Configuration;
using EasyZMq.Logging;
using NetMQ;
using NetMQ.Monitoring;

namespace EasyZMq.Sockets
{
    public abstract class EasyZMqReceiverSocket : IEasyZMqReceiverSocket
    {
        private readonly EasyZMqConfiguration _configuration;
        private readonly ILogger _logger;
        private NetMQContext _context;
        private NetMQSocket _socket;
        private NetMQMonitor _monitor;
        private Poller _poller;

        public event Action Connected;
        public event Action Disconnected;
        public event Action ConnectRetried;

        protected EasyZMqReceiverSocket(EasyZMqConfiguration configuration, NetMQContext context, NetMQSocket socket)
        {
            _configuration = configuration;
            _logger = configuration.EasyZMqLoggerFactory.GetLogger(typeof (EasyZMqReceiverSocket));
            _context = context;
            _socket = socket;

            CreatePoller(socket);
            CreateMonitor(context, socket, _poller);
            ConfigureSocket(socket);
        }

        public Uri Uri
        {
            get { return _configuration.AddressBinder.Uri; }
        }

        public void Start()
        {
            if (_disposedValue) throw new ObjectDisposedException("EasyZMqReceiverSocket");

            _configuration.AddressBinder.ConnectOrBindAddress(_socket);

            _poller.PollTillCancelledNonBlocking();
        }

        public abstract void OnMessageReceived<T>(T message);

        private void CreatePoller(ISocketPollable socket)
        {
            _poller = new Poller(socket);
        }

        private void CreateMonitor(NetMQContext context, NetMQSocket socket, Poller poller)
        {
            _monitor = new NetMQMonitor(context, socket, string.Format("inproc://{0}.inproc", Guid.NewGuid()),
                SocketEvents.Connected | SocketEvents.Disconnected | SocketEvents.ConnectRetried);

            _monitor.Connected += Monitor_Connected;
            _monitor.Disconnected += Monitor_Disconnected;
            _monitor.ConnectRetried += Monitor_ConnectRetried;

            _monitor.AttachToPoller(poller);
        }

        private void ConfigureSocket(NetMQSocket socket)
        {
            socket.Options.TcpKeepalive = true;
            socket.Options.TcpKeepaliveIdle = TimeSpan.FromSeconds(5);
            socket.Options.TcpKeepaliveInterval = TimeSpan.FromSeconds(1);
            socket.ReceiveReady += Subscriber_ReceiveReady;
        }

        private void Subscriber_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            bool more;
            e.Socket.ReceiveFrameString(out more);

            if (!more)
            {
                _logger.Warning("Invalid message received! The message was discarded.");
                return;
            }

            var bytes = e.Socket.ReceiveFrameBytes();

            DeserializeAndDispatch(bytes);
        }

        private void Monitor_Connected(object sender, NetMQMonitorSocketEventArgs e)
        {
            var handler = Connected;
            if (handler != null)
            {
                handler();
            }
        }

        private void Monitor_Disconnected(object sender, NetMQMonitorSocketEventArgs e)
        {
            var handler = Disconnected;
            if (handler != null)
            {
                handler();
            }
        }

        private void Monitor_ConnectRetried(object sender, NetMQMonitorIntervalEventArgs e)
        {
            var handler = ConnectRetried;
            if (handler != null)
            {
                handler();
            }
        }

        private void DeserializeAndDispatch(byte[] bytes)
        {
            try
            {
                dynamic message = _configuration.Serializer.Deserialize(bytes);
                OnMessageReceived(message);
            }
            catch (Exception ex)
            {
                _logger.Error("Deserialization and dispatch of incoming message failed.", ex);
            }
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        private bool _disposedValue;
        protected virtual void Dispose(bool disposing)
        {
            if (!_disposedValue)
            {
                if (disposing)
                {
                    if (_poller != null)
                    {
                        _poller.CancelAndJoin();

                        DisposeMonitor();
                        DisposePoller();
                        DisposeSocket();
                        DisposeContext();
                    }
                }

                _disposedValue = true;
            }
        }

        private void DisposeMonitor()
        {
            _monitor.Connected -= Monitor_Connected;
            _monitor.Disconnected -= Monitor_Disconnected;
            _monitor.ConnectRetried -= Monitor_ConnectRetried;
            _monitor.Dispose();
            _monitor = null;
        }

        private void DisposePoller()
        {
            _poller.Dispose();
            _poller = null;
        }

        private void DisposeSocket()
        {
            _socket.ReceiveReady -= Subscriber_ReceiveReady;
            _socket.Dispose();
            _socket = null;
        }
{'fraction_non_alphanumeric': 0.06935875864647598, 'fraction_numerical': 0.0005608524957936063, 'mean_word_length': 1.6319803198031981, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '1081494', 'n_tokens_mistral': 2678, 'n_tokens_neox': 2509, 'n_words': 549}