<commit_before>from zyre import Zyre, ZyreEvent
import zmq
import logging
from czmq import Zsock, string_at, Zmsg
import names
import uuid
import sys
from zmq.eventloop import ioloop
from argparse import ArgumentParser
from pyzyre.client import Client
from pyzyre.utils import resolve_endpoint
from pyzyre.constants import ZYRE_GROUP, LOG_FORMAT, SERVICE_PORT
import os
from pyzyre import color

logger = logging.getLogger('')

EVASIVE_TIMEOUT = os.environ.get('ZYRE_EVASIVE_TIMEOUT', 5000)  # zyre defaults
EXPIRED_TIMEOUT = os.environ.get('ZYRE_EXPIRED_TIMEOUT', 30000)


def task(pipe, arg):
    args = string_at(arg)
    args = dict(item.split("=") for item in args.split(","))

    name = args.get('name', names.get_full_name())
    group = args.get('group', ZYRE_GROUP)
    verbose = args.get('verbose')

    logger.info('setting up node: %s' % name)
    n = Zyre(name)

    logger.debug('setting evasive timeout: {}'.format(EVASIVE_TIMEOUT))
    n.set_evasive_timeout(int(EVASIVE_TIMEOUT))

    logger.debug('setting experation timer: {}'.format(EXPIRED_TIMEOUT))
    n.set_expired_timeout(int(EXPIRED_TIMEOUT))

    if verbose == '1':
        logger.info('setting verbose...')
        logger.setLevel(logging.DEBUG)
        n.set_verbose()

    if args.get('endpoint'):
        logger.info('setting endpoint: {}'.format(args['endpoint']))
        n.set_endpoint(args['endpoint'])

    if not args.get('beacon'):
        logger.info('setting up gossip')
        if args.get('gossip_bind'):
            logger.info('binding gossip: {}'.format(args['gossip_bind']))
            n.gossip_bind(args['gossip_bind'])
        else:
            logger.info('connecting to gossip group: {}'.format(args['gossip_connect']))
            n.gossip_connect(args['gossip_connect'])

    poller = zmq.Poller()

    pipe_zsock_s = Zsock(pipe, False)
    assert pipe_zsock_s

    pipe_s = zmq.Socket(shadow=pipe_zsock_s.resolve(pipe_zsock_s).value)
    assert pipe_s

    poller.register(pipe_s, zmq.POLLIN)

    # getting the zyre [czmq] socket
    # https://marc.info/?l=zeromq-dev&m=144719752222833&w=2
    s = n.socket()
    assert s

    # re-casting as native pyzmq socket
    ss = zmq.Socket(shadow=s.resolve(s).value)
    assert ss

    # registering
    poller.register(ss, zmq.POLLIN)

    logger.info('staring node...')
    n.start()

    group = group.split('|')
    for g in group:
        logger.info('joining: %s' % g)
        n.join(g)

    pipe_zsock_s.signal(0)  # OK

    peers = {}
    terminated = False
    while not terminated:
        items = dict(poller.poll())

        try:
            if pipe_s in items and items[pipe_s] == zmq.POLLIN:
                message = Zmsg.recv(pipe)

                if not message:
                    break  # SIGINT

                msg_type = message.popstr().decode('utf-8')

                # message to quit
                if msg_type == "$$STOP":
                    terminated = True
                elif msg_type == '$$ID':
                    pipe_s.send_string(n.uuid().decode('utf-8'))
                else:
                    if msg_type == 'whisper':
                        address = message.popstr().decode('utf-8')
                        msg = message.popstr().decode('utf-8')
                        m = Zmsg()
                        m.addstr(msg)
                        address = peers[str(address)]
                        address = uuid.UUID(address).hex.upper()
                        n.whisper(address, m)
                    else:
                        msg = message.popstr().decode('utf-8')
                        n.shouts(group, msg.encode('utf-8'))

            elif ss in items and items[ss] == zmq.POLLIN:
                e = ZyreEvent(n)

                msg_type = e.type().decode('utf-8')
                logger.debug('found ZyreEvent: %s' % msg_type)

                if msg_type == "ENTER":
                    logger.debug('ENTER {} - {}'.format(e.peer_name(), e.peer_uuid()))
                    peers[e.peer_name()] = e.peer_uuid()
                    pipe_s.send_multipart(['ENTER', e.peer_uuid(), e.peer_name()])
                    #headers = e.headers() # zlist<commit_msg>bugfix to shouts for default group<commit_after>from zyre import Zyre, ZyreEvent
import zmq
import logging
from czmq import Zsock, string_at, Zmsg
import names
import uuid
import sys
from zmq.eventloop import ioloop
from argparse import ArgumentParser
from pyzyre.client import Client
from pyzyre.utils import resolve_endpoint
from pyzyre.constants import ZYRE_GROUP, LOG_FORMAT, SERVICE_PORT
import os
from pyzyre import color

logger = logging.getLogger('')

EVASIVE_TIMEOUT = os.environ.get('ZYRE_EVASIVE_TIMEOUT', 5000)  # zyre defaults
EXPIRED_TIMEOUT = os.environ.get('ZYRE_EXPIRED_TIMEOUT', 30000)


def task(pipe, arg):
    args = string_at(arg)
    args = dict(item.split("=") for item in args.split(","))

    name = args.get('name', names.get_full_name())
    group = args.get('group', ZYRE_GROUP)
    verbose = args.get('verbose')

    logger.info('setting up node: %s' % name)
    n = Zyre(name)

    logger.debug('setting evasive timeout: {}'.format(EVASIVE_TIMEOUT))
    n.set_evasive_timeout(int(EVASIVE_TIMEOUT))

    logger.debug('setting experation timer: {}'.format(EXPIRED_TIMEOUT))
    n.set_expired_timeout(int(EXPIRED_TIMEOUT))

    if verbose == '1':
        logger.info('setting verbose...')
        logger.setLevel(logging.DEBUG)
        n.set_verbose()

    if args.get('endpoint'):
        logger.info('setting endpoint: {}'.format(args['endpoint']))
        n.set_endpoint(args['endpoint'])

    if not args.get('beacon'):
        logger.info('setting up gossip')
        if args.get('gossip_bind'):
            logger.info('binding gossip: {}'.format(args['gossip_bind']))
            n.gossip_bind(args['gossip_bind'])
        else:
            logger.info('connecting to gossip group: {}'.format(args['gossip_connect']))
            n.gossip_connect(args['gossip_connect'])

    poller = zmq.Poller()

    pipe_zsock_s = Zsock(pipe, False)
    assert pipe_zsock_s

    pipe_s = zmq.Socket(shadow=pipe_zsock_s.resolve(pipe_zsock_s).value)
    assert pipe_s

    poller.register(pipe_s, zmq.POLLIN)

    # getting the zyre [czmq] socket
    # https://marc.info/?l=zeromq-dev&m=144719752222833&w=2
    s = n.socket()
    assert s

    # re-casting as native pyzmq socket
    ss = zmq.Socket(shadow=s.resolve(s).value)
    assert ss

    # registering
    poller.register(ss, zmq.POLLIN)

    logger.info('staring node...')
    n.start()

    group = group.split('|')
    for g in group:
        logger.info('joining: %s' % g)
        n.join(g)

    pipe_zsock_s.signal(0)  # OK

    peers = {}
    terminated = False
    while not terminated:
        items = dict(poller.poll())

        try:
            if pipe_s in items and items[pipe_s] == zmq.POLLIN:
                message = Zmsg.recv(pipe)

                if not message:
                    break  # SIGINT

                msg_type = message.popstr().decode('utf-8')

                # message to quit
                if msg_type == "$$STOP":
                    terminated = True
                elif msg_type == '$$ID':
                    pipe_s.send_string(n.uuid().decode('utf-8'))
                else:
                    if msg_type == 'whisper':
                        address = message.popstr().decode('utf-8')
                        msg = message.popstr().decode('utf-8')
                        m = Zmsg()
                        m.addstr(msg)
                        address = peers[str(address)]
                        address = uuid.UUID(address).hex.upper()
                        n.whisper(address, m)
                    else:
                        msg = message.popstr().decode('utf-8')
                        n.shouts(group[0], msg.encode('utf-8'))

            elif ss in items and items[ss] == zmq.POLLIN:
                e = ZyreEvent(n)

                msg_type = e.type().decode('utf-8')
                logger.debug('found ZyreEvent: %s' % msg_type)

                if msg_type == "ENTER":
                    logger.debug('ENTER {} - {}'.format(e.peer_name(), e.peer_uuid()))
                    peers[e.peer_name()] = e.peer_uuid()
                    pipe_s.send_multipart(['ENTER', e.peer_uuid(), e.peer_name()])
                    #headers = e.headers() # zlist
{'fraction_non_alphanumeric': 0.1293639406982305, 'fraction_numerical': 0.00824964131994261, 'mean_word_length': 2.0935650887573964, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, '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': '5345453', 'n_tokens_mistral': 2730, 'n_tokens_neox': 2594, 'n_words': 534}