<commit_before>﻿using NetMQ;
using PortForward.Utilities;
using System;
using System.Threading.Tasks;

namespace PortForward.NetworkBridge
{
    public class ZMQClient : Client
    {
        private NetMQContext _context;
        private NetMQSocket _pubSocket;
        private NetMQSocket _subSocket;
        private string _topic = "raw-stream";
        private string _remote;

        public ZMQClient(string publisherAddress)
        {
            _remote = string.Format("tcp://{0}:4040", publisherAddress);
            
            _context = NetMQContext.Create();
            _pubSocket = _context.CreatePublisherSocket();
            _subSocket = _context.CreateSubscriberSocket();

            _pubSocket.Bind("tcp://*:4040");
            Task.Factory.StartNew(ListenerThread);
        }

        public override void HandleResponse(object sender, EventArgs e)
        {
            byte[] bytes = (byte[])sender;
            _pubSocket.SendMore(_topic).Send(bytes);
        }

        private void ListenerThread()
        {
            int minReconnectDelay = 1;
            int reconnectDelay = minReconnectDelay;
            int maxReconnectDelay = 60;

            while (true)
            {
                try
                {
                    _subSocket.Bind(_remote);
                    reconnectDelay = minReconnectDelay;

                    while (true)
                    {
                        string response = _subSocket.ReceiveString();
                        byte[] bytes = ByteStringConverter.GetBytes(response);

                        Push(bytes);
                    }
                }
                catch (Exception e)<commit_msg>tested: can now send between remote clients<commit_after>﻿using NetMQ;
using PortForward.Utilities;
using System;
using System.Threading.Tasks;

namespace PortForward.NetworkBridge
{
    public class ZMQClient : Client
    {
        private NetMQContext _context;
        private NetMQSocket _pubSocket;
        private NetMQSocket _subSocket;
        private string _topic = "raw-stream";
        private string _remote;

        public ZMQClient(string publisherAddress)
        {
            _remote = string.Format("tcp://{0}:4040", publisherAddress);
            
            _context = NetMQContext.Create();
            _pubSocket = _context.CreatePublisherSocket();
            _subSocket = _context.CreateSubscriberSocket();

            _pubSocket.Bind("tcp://*:4040");
            Task.Factory.StartNew(ListenerThread);
        }

        public override void HandleResponse(object sender, EventArgs e)
        {
            byte[] bytes = (byte[])sender;
            _pubSocket.SendMore(_topic).Send(bytes, bytes.Length, dontWait: true);
        }

        private void ListenerThread()
        {
            int minReconnectDelay = 1;
            int reconnectDelay = minReconnectDelay;
            int maxReconnectDelay = 60;

            while (true)
            {
                try
                {
                    _subSocket.Connect(_remote);
                    _subSocket.Subscribe(_topic);
                    reconnectDelay = minReconnectDelay;

                    while (true)
                    {
                        string response = _subSocket.ReceiveString();
                        byte[] bytes = ByteStringConverter.GetBytes(response);

                        Push(bytes);
                    }
                }
                catch (Exception e)
{'fraction_non_alphanumeric': 0.07985010089362929, 'fraction_numerical': 0.006918420294032862, 'mean_word_length': 1.5837676842889055, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1197893', 'n_tokens_mistral': 878, 'n_tokens_neox': 801, 'n_words': 179}