<commit_before>#!/usr/bin/env python
"""
Command-line interface for various Zeeko components
"""

import click
import logging
import zmq
import time
import sys
import numpy as np

from zutils import zmain, ioloop, MessageLine


log = logging.getLogger()

main = zmain()

@main.command()
@click.option("--interval", type=int, help="Polling interval for client status.", default=3)
@click.pass_context
def proxy(ctx, interval):
    """A proxy object for monitoring traffic between two sockets"""
    xpub = ctx.obj.zcontext.socket(zmq.XPUB)
    xsub = ctx.obj.zcontext.socket(zmq.XSUB)
    
    xpub.bind(ctx.obj.secondary.bind)
    xsub.bind(ctx.obj.primary.bind)
    
    click.echo("XPUB at {0}".format(ctx.obj.secondary.bind))
    click.echo("XSUB at {0}".format(ctx.obj.primary.bind))
    
    poller = zmq.Poller()
    poller.register(xpub, zmq.POLLIN)
    poller.register(xsub, zmq.POLLIN)
    
    rate = 0.0
    last_message = time.time()
    
    while True:
        start = time.time()
        while last_message + interval > start:
            data = 0
            sockets = dict(poller.poll(timeout=10))
            if sockets.get(xpub, 0) & zmq.POLLIN:
                msg = xpub.recv_multipart()
                click.echo("[BROKER]: SUB {0!r}".format(msg))
                xsub.send_multipart(msg)
                data += sum(len(m) for m in msg)
            if sockets.get(xsub, 0) & zmq.POLLIN:
                msg = xsub.recv_multipart()
                xpub.send_multipart(msg)
                data += sum(len(m) for m in msg)
            end = time.time()
            rate = (rate * 0.9) + (data / (end - start)) * 0.1
            start = time.time()
        click.echo("Rate = {:.2f} Mb/s".format(rate / (1024 ** 2)))
        last_message = time.time()

@main.command()
@click.option("--interval", type=int, help="Polling interval for client status.", default=3)
@click.option("--subscribe", type=str, default="", help="Subscription value.")
@click.pass_context
def client(ctx, interval, subscribe):
    """Make a client"""<commit_msg>CLI proxy parses PUB/SUB info messages<commit_after>#!/usr/bin/env python
"""
Command-line interface for various Zeeko components
"""

import click
import logging
import zmq
import time
import sys
import numpy as np

from zutils import zmain, ioloop, MessageLine


log = logging.getLogger()

main = zmain()

@main.command()
@click.option("--interval", type=int, help="Polling interval for client status.", default=3)
@click.pass_context
def proxy(ctx, interval):
    """A proxy object for monitoring traffic between two sockets"""
    xpub = ctx.obj.zcontext.socket(zmq.XPUB)
    xsub = ctx.obj.zcontext.socket(zmq.XSUB)
    
    xpub.bind(ctx.obj.secondary.bind)
    xsub.bind(ctx.obj.primary.bind)
    
    click.echo("XPUB at {0}".format(ctx.obj.secondary.bind))
    click.echo("XSUB at {0}".format(ctx.obj.primary.bind))
    
    poller = zmq.Poller()
    poller.register(xpub, zmq.POLLIN)
    poller.register(xsub, zmq.POLLIN)
    
    rate = 0.0
    last_message = time.time()
    
    while True:
        start = time.time()
        while last_message + interval > start:
            data = 0
            sockets = dict(poller.poll(timeout=10))
            if sockets.get(xpub, 0) & zmq.POLLIN:
                msg = xpub.recv_multipart()
                if len(msg) == 1 and msg[0][0] in "\x00\x01":
                    if msg[0][0] == '\x00':
                        click.echo("[BROKER]: unsubscribe '{0}'".format(msg[0][1:]))
                    elif msg[0][0] == '\x01':
                        click.echo("[BROKER]: subscribe '{0}'".format(msg[0][1:]))
                xsub.send_multipart(msg)
                data += sum(len(m) for m in msg)
            if sockets.get(xsub, 0) & zmq.POLLIN:
                msg = xsub.recv_multipart()
                xpub.send_multipart(msg)
                data += sum(len(m) for m in msg)
            end = time.time()
            rate = (rate * 0.9) + (data / (end - start)) * 0.1
            start = time.time()
        click.echo("Rate = {:.2f} Mb/s".format(rate / (1024 ** 2)))
        last_message = time.time()

@main.command()
@click.option("--interval", type=int, help="Polling interval for client status.", default=3)
@click.option("--subscribe", type=str, default="", help="Subscription value.")
@click.pass_context
def client(ctx, interval, subscribe):
    """Make a client"""
{'fraction_non_alphanumeric': 0.1488732073753699, 'fraction_numerical': 0.014568631914409288, 'mean_word_length': 2.4221183800623054, 'pattern_counts': {'":': 1, '<': 3, '<?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': '5652022', 'n_tokens_mistral': 1500, 'n_tokens_neox': 1441, 'n_words': 334}