<commit_before># Copyright 2002-2003 <NAME>.  See LICENSE for licensing information.
# $Id: ServerMain.py,v 1.24 2003/01/07 19:17:57 nickm Exp $

"""mixminion.ServerMain

   The main loop and related functionality for a Mixminion server.

   See the "MixminionServer" class for more information about how it
   all works. """

__all__ = [ 'MixminonServer' ]

import fcntl
import getopt
import os
import sys
import time

import mixminion.Config
import mixminion.Crypto
import mixminion.server.MMTPServer
import mixminion.server.Modules
import mixminion.server.PacketHandler
import mixminion.server.Queue
import mixminion.server.ServerConfig
import mixminion.server.ServerKeys

from bisect import insort
from mixminion.Common import LOG, LogStream, MixError, MixFatalError, ceilDiv,\
     createPrivateDir, formatBase64, formatTime, waitForChildren

class IncomingQueue(mixminion.server.Queue.DeliveryQueue):
    """A DeliveryQueue to accept messages from incoming MMTP connections,
       process them with a packet handler, and send them into a mix pool."""

    def __init__(self, location, packetHandler):
        """Create an IncomingQueue that stores its messages in <location>
           and processes them through <packetHandler>."""
        mixminion.server.Queue.DeliveryQueue.__init__(self, location)
        self.packetHandler = packetHandler
        self.mixPool = None

    def connectQueues(self, mixPool):
        """Sets the target mix queue"""
        self.mixPool = mixPool

    def queueMessage(self, msg):
        """Add a message for delivery"""
        LOG.trace("Inserted message %s into incoming queue",
                  formatBase64(msg[:8]))
        self.queueDeliveryMessage(None, msg)

    def _deliverMessages(self, msgList):
        "Implementation of abstract method from DeliveryQueue."
        ph = self.packetHandler
        for handle, _, message, n_retries in msgList:
            try:
                res = ph.processMessage(message)
                if res is None:
                    # Drop padding before it gets to the mix.
                    LOG.debug("Padding message %s dropped",
                              formatBase64(message[:8]))
                    self.deliverySucceeded(handle)
                else:
                    LOG.debug("Processed message %s; inserting into pool",
                              formatBase64(message[:8]))
                    self.mixPool.queueObject(res)
                    self.deliverySucceeded(handle)
            except mixminion.Crypto.CryptoError, e:
                LOG.warn("Invalid PK or misencrypted packet header: %s", e)
                self.deliveryFailed(handle)
            except mixminion.Packet.ParseError, e:
                LOG.warn("Malformed message dropped: %s", e)
                self.deliveryFailed(handle)
            except mixminion.server.PacketHandler.ContentError, e:
                LOG.warn("Discarding bad packet: %s", e)
                self.deliveryFailed(handle)

class MixPool:
    """Wraps a mixminion.server.Queue.*MixQueue to send messages to an exit
       queue and a delivery queue."""
    def __init__(self, config, queueDir):
        """Create a new MixPool, based on this server's configuration and
           queue location."""

        server = config['Server']
        interval = server['MixInterval'][2]
        if server['MixAlgorithm'] == 'TimedMixQueue':
            self.queue = mixminion.server.Queue.TimedMixQueue(
                location=queueDir, interval=interval)
        elif server['MixAlgorithm'] == 'CottrellMixQueue':
            self.queue = mixminion.server.Queue.CottrellMixQueue(
                location=queueDir, interval=interval,
                minPool=server.get("MixPoolMinSize", 5),
                sendRate=server.get("MixPoolRate", 0.6))
        elif server['MixAlgorithm'] == 'BinomialCottrellMixQueue':
            self.queue = mixminion.server.Queue.BinomialCottrellMixQueue(
                location=queueDir, interval=interval,
                minPool=server.get("MixPoolMinSize", 5),
                sendRate=server.get("MixPoolRate", 0.6))
        else:
            raise MixFatalError("Got impossible mix queue type from config")

        self.outgoingQueue = None
        self.moduleManager = None

    def queueObject(self, obj):
        """Insert an object into the queue."""
        self.queue.queueObject(obj)

    def count(self):
        "Return the number of messages in the queue"
        return self.queue.count()

    def connectQueues(self, outgoing, manager):
        """Sets the queue for outgoing mixminion packets, and the
           module manager for deliverable messages."""
        self.outgoingQueue = outgoing
        self.moduleManager = manager

    def mix(self):
        """Get a batch of messages, and queue them for delivery as
           appropriate."""
        if self.queue.count() == 0:
            LOG.trace("No messages in the mix pool")
            return
        handles = self.queue.getBatch()
        LOG.debug("%s messages in the mix pool; delivering %s.",
                  self.queue.count(), len(handles)
        
        for h in handles:
            tp, info = self.queue.getObject(h)
            if tp == 'EXIT':
                rt, ri, app_key, tag, payload = info
                LOG.debug("  (sending message %s to exit modules)",<commit_msg>fix syntax error from last patch<commit_after># Copyright 2002-2003 <NAME>.  See LICENSE for licensing information.
# $Id: ServerMain.py,v 1.25 2003/01/07 19:19:54 nickm Exp $

"""mixminion.ServerMain

   The main loop and related functionality for a Mixminion server.

   See the "MixminionServer" class for more information about how it
   all works. """

__all__ = [ 'MixminonServer' ]

import fcntl
import getopt
import os
import sys
import time

import mixminion.Config
import mixminion.Crypto
import mixminion.server.MMTPServer
import mixminion.server.Modules
import mixminion.server.PacketHandler
import mixminion.server.Queue
import mixminion.server.ServerConfig
import mixminion.server.ServerKeys

from bisect import insort
from mixminion.Common import LOG, LogStream, MixError, MixFatalError, ceilDiv,\
     createPrivateDir, formatBase64, formatTime, waitForChildren

class IncomingQueue(mixminion.server.Queue.DeliveryQueue):
    """A DeliveryQueue to accept messages from incoming MMTP connections,
       process them with a packet handler, and send them into a mix pool."""

    def __init__(self, location, packetHandler):
        """Create an IncomingQueue that stores its messages in <location>
           and processes them through <packetHandler>."""
        mixminion.server.Queue.DeliveryQueue.__init__(self, location)
        self.packetHandler = packetHandler
        self.mixPool = None

    def connectQueues(self, mixPool):
        """Sets the target mix queue"""
        self.mixPool = mixPool

    def queueMessage(self, msg):
        """Add a message for delivery"""
        LOG.trace("Inserted message %s into incoming queue",
                  formatBase64(msg[:8]))
        self.queueDeliveryMessage(None, msg)

    def _deliverMessages(self, msgList):
        "Implementation of abstract method from DeliveryQueue."
        ph = self.packetHandler
        for handle, _, message, n_retries in msgList:
            try:
                res = ph.processMessage(message)
                if res is None:
                    # Drop padding before it gets to the mix.
                    LOG.debug("Padding message %s dropped",
                              formatBase64(message[:8]))
                    self.deliverySucceeded(handle)
                else:
                    LOG.debug("Processed message %s; inserting into pool",
                              formatBase64(message[:8]))
                    self.mixPool.queueObject(res)
                    self.deliverySucceeded(handle)
            except mixminion.Crypto.CryptoError, e:
                LOG.warn("Invalid PK or misencrypted packet header: %s", e)
                self.deliveryFailed(handle)
            except mixminion.Packet.ParseError, e:
                LOG.warn("Malformed message dropped: %s", e)
                self.deliveryFailed(handle)
            except mixminion.server.PacketHandler.ContentError, e:
                LOG.warn("Discarding bad packet: %s", e)
                self.deliveryFailed(handle)

class MixPool:
    """Wraps a mixminion.server.Queue.*MixQueue to send messages to an exit
       queue and a delivery queue."""
    def __init__(self, config, queueDir):
        """Create a new MixPool, based on this server's configuration and
           queue location."""

        server = config['Server']
        interval = server['MixInterval'][2]
        if server['MixAlgorithm'] == 'TimedMixQueue':
            self.queue = mixminion.server.Queue.TimedMixQueue(
                location=queueDir, interval=interval)
        elif server['MixAlgorithm'] == 'CottrellMixQueue':
            self.queue = mixminion.server.Queue.CottrellMixQueue(
                location=queueDir, interval=interval,
                minPool=server.get("MixPoolMinSize", 5),
                sendRate=server.get("MixPoolRate", 0.6))
        elif server['MixAlgorithm'] == 'BinomialCottrellMixQueue':
            self.queue = mixminion.server.Queue.BinomialCottrellMixQueue(
                location=queueDir, interval=interval,
                minPool=server.get("MixPoolMinSize", 5),
                sendRate=server.get("MixPoolRate", 0.6))
        else:
            raise MixFatalError("Got impossible mix queue type from config")

        self.outgoingQueue = None
        self.moduleManager = None

    def queueObject(self, obj):
        """Insert an object into the queue."""
        self.queue.queueObject(obj)

    def count(self):
        "Return the number of messages in the queue"
        return self.queue.count()

    def connectQueues(self, outgoing, manager):
        """Sets the queue for outgoing mixminion packets, and the
           module manager for deliverable messages."""
        self.outgoingQueue = outgoing
        self.moduleManager = manager

    def mix(self):
        """Get a batch of messages, and queue them for delivery as
           appropriate."""
        if self.queue.count() == 0:
            LOG.trace("No messages in the mix pool")
            return
        handles = self.queue.getBatch()
        LOG.debug("%s messages in the mix pool; delivering %s.",
                  self.queue.count(), len(handles))
        
        for h in handles:
            tp, info = self.queue.getObject(h)
            if tp == 'EXIT':
                rt, ri, app_key, tag, payload = info
                LOG.debug("  (sending message %s to exit modules)",
{'fraction_non_alphanumeric': 0.09147589800856132, 'fraction_numerical': 0.008189093616229295, 'mean_word_length': 2.3251856435643563, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '2744498', 'n_tokens_mistral': 2926, 'n_tokens_neox': 2771, 'n_words': 838}