<commit_before>#!/usr/bin/env python

# yet another attempt at a proper UnixProcess able to cope with buffer limiting
# both on input to the subprocess and on output to the destination inbox

from Axon.Component import component
from Axon.Ipc import shutdownMicroprocess, producerFinished
from Axon.AxonExceptions import noSpaceInBox
from Kamaelia.IPC import newReader, newWriter
from Kamaelia.IPC import removeReader, removeWriter

from Kamaelia.Internet.Selector import Selector

import subprocess
import fcntl
import os
import sys


class UnixProcess(component):

    Outboxes = { "outbox" : "",
                 "error"  : "",
                 "signal" : "",
               }

    def __init__(self, command, buffersize=32768):
        super(UnixProcess,self).__init__()
        self.command = command
        self.buffersize = buffersize

    def main(self):
        # set up the subprocess
        p = subprocess.Popen(self.command, 
                             shell=True, 
                             bufsize=self.buffersize, 
                             stdin=subprocess.PIPE, 
                             stdout=subprocess.PIPE, 
                             stderr = subprocess.PIPE, 
                             close_fds=True)

        makeNonBlocking( p.stdin.fileno() )
        makeNonBlocking( p.stdout.fileno() )
        makeNonBlocking( p.stderr.fileno() )
        
        # set up child components to handle the IO
        
        STDIN = _ToFileHandle(p.stdin)
        STDOUT = _FromFileHandle(p.stdout, self.buffersize)
        STDERR = _FromFileHandle(p.stderr, self.buffersize)

        linkages = [
            # stdin from inbox; stdout and stderr to outboxes
            self.link((self,"inbox"),    (STDIN,"inbox"), passthrough=1),
            self.link((STDOUT,"outbox"), (self,"outbox"), passthrough=2),
            self.link((STDERR,"outbox"), (self,"error"),  passthrough=2),

            # if outputs close, then close input too
            self.link((STDOUT,"signal"), (STDIN,"control")),
            self.link((STDERR,"signal"), (STDIN,"control")),

            # if ordered from outside, then close input
            self.link((self,"control"), (STDIN, "control"), passthrough=1),
        ]

        shutdownMsg = producerFinished(self)
        
        STDIN.activate()
        STDOUT.activate()
        STDERR.activate()
        self.addChildren(STDIN,STDOUT,STDERR)

        while not self.childrenDone():
            self.pause()
            yield 1

        self.send(shutdownMsg,"signal")

    def childrenDone(self):
        """Unplugs any children that have terminated, and returns true if there are no
           running child components left (ie. their microproceses have finished)
        """
        for child in self.childComponents():<commit_msg>Added to UnixProcess ability to instantiate named pipes to provide additional channels for data.<commit_after>#!/usr/bin/env python

# yet another attempt at a proper UnixProcess able to cope with buffer limiting
# both on input to the subprocess and on output to the destination inbox

from Axon.Component import component
from Axon.Ipc import shutdownMicroprocess, producerFinished
from Axon.AxonExceptions import noSpaceInBox
from Kamaelia.IPC import newReader, newWriter
from Kamaelia.IPC import removeReader, removeWriter

from Kamaelia.Internet.Selector import Selector

import subprocess
import fcntl
import os
import sys


class UnixProcess(component):

    Outboxes = { "outbox" : "",
                 "error"  : "",
                 "signal" : "",
                 "_shutdownOutpipes" : "For shutting down any named pipes used for output"
               }

    def __init__(self, command, buffersize=32768, outpipes={}):
        for outpipe,outboxname in outpipes.items():
            self.Outboxes[outboxname] = "Output from named pipe: "+outpipe
            
        super(UnixProcess,self).__init__()
        self.command = command
        self.buffersize = buffersize
        self.outpipes = outpipes

    def main(self):
        # lets add any named output pipes requested
        outpipeshutdown=(self,"_shutdownOutpipes")
        for (outpipename,outboxname) in self.outpipes.items():
            os.mkfifo(outpipename)
            
            f = open(outpipename, "rb+",self.buffersize)
            makeNonBlocking(f)
            
            PIPE = _FromFileHandle(f, self.buffersize)
            self.link((PIPE,"outbox"), (self,outboxname), passthrough=2)
            # shutdown chain
            self.link(outpipeshutdown,(PIPE,"control"))
            outpipeshutdown=(PIPE,"signal")
            
            self.addChildren(PIPE)
            
        # set up the subprocess
        p = subprocess.Popen(self.command, 
                             shell=True, 
                             bufsize=self.buffersize, 
                             stdin=subprocess.PIPE, 
                             stdout=subprocess.PIPE, 
                             stderr = subprocess.PIPE, 
                             close_fds=True)

        makeNonBlocking( p.stdin.fileno() )
        makeNonBlocking( p.stdout.fileno() )
        makeNonBlocking( p.stderr.fileno() )
        
        # set up child components to handle the IO
        
        STDIN = _ToFileHandle(p.stdin)
        STDOUT = _FromFileHandle(p.stdout, self.buffersize)
        STDERR = _FromFileHandle(p.stderr, self.buffersize)

        # stdin from inbox; stdout and stderr to outboxes
        self.link((self,"inbox"),    (STDIN,"inbox"), passthrough=1),
        self.link((STDOUT,"outbox"), (self,"outbox"), passthrough=2),
        self.link((STDERR,"outbox"), (self,"error"),  passthrough=2),

        # if outputs close, then close input too
        self.link((STDOUT,"signal"), (STDIN,"control")),
        self.link((STDERR,"signal"), (STDIN,"control")),

        # if ordered from outside, then close input
        self.link((self,"control"), (STDIN, "control"), passthrough=1),

        self.addChildren(STDIN,STDOUT,STDERR)
            
        for child in self.childComponents():
            child.activate()

        shutdownMsg = producerFinished(self)
        
        while not self.childrenDone():
            if p.poll() is not None:
                if self.outpipes:
                    self.send(shutdownMicroprocess(self), "_shutdownOutpipes")
            self.pause()
            yield 1

        self.send(shutdownMsg,"signal")
        
        # delete any named pipes
        for outpipename in self.outpipes.keys():
            os.remove(outpipename)
        

    def childrenDone(self):
        """Unplugs any children that have terminated, and returns true if there are no
           running child components left (ie. their microproceses have finished)
        """
        for child in self.childComponents():
{'fraction_non_alphanumeric': 0.09641025641025641, 'fraction_numerical': 0.003076923076923077, 'mean_word_length': 2.1041382446566623, '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': '2352501', 'n_tokens_mistral': 1976, 'n_tokens_neox': 1860, 'n_words': 474}