Processes files that have been added from the rq-worker, starts plugins
    that match the mime type for the new file.
186421
def file_queue(path, template_path='/vent/', r_host='redis'):
    
    import configparser
    import ast
    import docker
    import json
    import requests
    import os
    import sys
    import time
    import uuid

    from redis import Redis
    from rq import Queue
    from subprocess import check_output, Popen, PIPE
    from string import punctuation
    from vent.helpers.logs import Logger

    status = (True, None)
    images = []
    configs = {}
    logger = Logger(__name__)

    if (os.path.isfile('/root/.vent/vent.cfg') and os.path.isfile('/root/.vent/plugin_manifest.cfg')):
        template_path = '/root/.vent/'

    try:
        d_client = docker.from_env()

        # get the correct path for binding
        vent_config = configparser.ConfigParser(interpolation=None)
        vent_config.optionxform = str
        vent_config.read(template_path+'vent.cfg')
        if (vent_config.has_section('main') and
                vent_config.has_option('main', 'files')):
            files = vent_config.get('main', 'files')
        else:
            files = '/'

        # deal with ~
        files = os.path.expanduser(files)

        chars = set(punctuation)
        chars.discard('/')
        chars.discard('_')
        chars.discard('-')
        file_name = ''
        # escape any funky symbols to allow users FREEDOM of directory name
        for char in files:
            if char in chars:
                if char == '\\':
                    file_name += '\\' + char
                else:
                    file_name += '\\\\' + char
            else:
                file_name += char

        files = file_name
        _, path = path.split('_', 1)
        directory = path.rsplit('/', 1)[0]
        path = path.replace('/files', files, 1)
        path_copy = path

        # read in configuration of plugins to get the ones that should run
        # against the path.
        # keep track of images that failed getting configurations for
        failed_images = set()
        config = configparser.ConfigParser(interpolation=None)
        config.optionxform = str
        print('Path to manifest: ' + template_path+'plugin_manifest.cfg')
        config.read(template_path+'plugin_manifest.cfg')
        sections = config.sections()
        name_maps = {}
        orig_path_d = {}
        path_cmd = {}
        labels_d = {}

        # get all name maps
        for section in sections:
            link_name = config.get(section, 'link_name')
            image_name = config.get(section, 'image_name')
            name_maps[link_name] = image_name.replace(
                ':', '-').replace('/', '-')

        for section in sections:
            path = path_copy
            orig_path = ''
            repo = config.get(section, 'repo')
            t_type = config.get(section, 'type')
            labels = {'vent-plugin': '', 'file': path,
                      'vent.section': section, 'vent.repo': repo, 'vent.type': t_type}
            image_name = config.get(section, 'image_name')
            link_name = config.get(section, 'link_name')
            # doesn't matter if it's a repository or registry because both in manifest
            if config.has_option(section, 'groups'):
                if 'replay' in config.get(section, 'groups'):
                    try:
                        # read the vent.cfg file to grab the network-mapping
                        # specified. For replay_pcap
                        n_name = 'network-mapping'
                        n_map = []
                        if vent_config.has_section(n_name):
                            # make sure that the options aren't empty
                            if vent_config.options(n_name):
                                options = vent_config.options(n_name)
                                for option in options:
                                    if vent_config.get(n_name, option):
                                        n_map.append(vent_config.get(
                                            n_name, option))
                                orig_path = path
                                path = str(n_map[0]) + ' ' + path
                    except Exception as e:  # pragma: no cover
                        failed_images.add(image_name)
                        status = (False, str(e))
            if config.has_option(section, 'service'):
                try:
                    options_dict = json.loads(config.get(section, 'service'))
                    for option in options_dict:
                        value = options_dict[option]
                        labels[option] = value
                except Exception as e:   # pragma: no cover
                    failed_images.add(image_name)
                    status = (False, str(e))
            if config.has_option(section, 'settings'):
                try:
                    options_dict = json.loads(config.get(section, 'settings'))
                    in_base = directory == '/files'
                    # process base by default
                    process_file = in_base
                    # check if this tool shouldn't process the base by default
                    if 'process_base' in options_dict:
                        if options_dict['process_base'] == 'no':
                            process_file = False
                    # check if this tool should look at subdirs created by
                    # other tools' output
                    if 'process_from_tool' in options_dict and not in_base:
                        for tool in options_dict['process_from_tool'].split(','):
                            dir_pieces = directory.split('/')
                            dir_check = dir_pieces
                            for dir_piece in dir_pieces:
                                if 'UTC' in dir_piece:
                                    dir_check = dir_piece
                            if tool.replace(' ', '-') in dir_check:
                                process_file = True
                    if 'ext_types' in options_dict and process_file:
                        ext_types = options_dict['ext_types'].split(',')
                        for ext_type in ext_types:
                            if path.endswith(ext_type):
                                images.append(image_name)
                                configs[image_name] = {}

                except Exception as e:   # pragma: no cover
                    failed_images.add(image_name)
                    status = (False, str(e))
            if image_name in configs:
                if config.has_option(section, 'docker'):
                    try:
                        options_dict = ast.literal_eval(
                            config.get(section, 'docker'))
                        for option in options_dict:
                            try:
                                configs[image_name][option] = ast.literal_eval(
                                    options_dict[option])
                            except Exception as e:  # pragma: no cover
                                configs[image_name][option] = options_dict[option]
                        if 'links' in configs[image_name]:
                            for link in configs[image_name]['links']:
                                if link in name_maps:
                                    configs[image_name]['links'][name_maps[link]
                                                                 ] = configs[image_name]['links'].pop(link)
                        # TODO network_mode
                        # TODO volumes_from
                        # TODO external services
                    except Exception as e:   # pragma: no cover
                        failed_images.add(image_name)
                        status = (False, str(e))

            if config.has_option(section, 'gpu') and image_name in configs:
                try:
                    options_dict = json.loads(config.get(section, 'gpu'))
                    if 'enabled' in options_dict:
                        enabled = options_dict['enabled']
                        if enabled == 'yes':
                            configs[image_name]['gpu_options'] = options_dict
                            labels['vent.gpu'] = 'yes'
                            if 'dedicated' in options_dict:
                                labels['vent.gpu.dedicated'] = options_dict['dedicated']
                            if 'device' in options_dict:
                                labels['vent.gpu.device'] = options_dict['device']
                            if 'mem_mb' in options_dict:
                                labels['vent.gpu.mem_mb'] = options_dict['mem_mb']
                            port = ''
                            host = ''
                            if (vent_config.has_section('nvidia-docker-plugin') and
                                    vent_config.has_option('nvidia-docker-plugin', 'port')):
                                port = vent_config.get(
                                    'nvidia-docker-plugin', 'port')
                            else:
                                port = '3476'
                            if (vent_config.has_section('nvidia-docker-plugin') and
                                    vent_config.has_option('nvidia-docker-plugin', 'host')):
                                host = vent_config.get(
                                    'nvidia-docker-plugin', 'host')
                            else:
                                # grab the default gateway
                                try:
                                    route = Popen(('/sbin/ip', 'route'),
                                                  stdout=PIPE)
                                    host = check_output(('awk', '/default/ {print$3}'),
                                                        stdin=route.stdout).strip().decode('utf-8')
                                    route.wait()
                                except Exception as e:  # pragma no cover
                                    logger.error('Default gateway '
                                                 'went wrong ' + str(e))
                            nd_url = 'http://' + host + ':' + port + '/v1.0/docker/cli'
                            params = {'vol': 'nvidia_driver'}
                            try:
                                r = requests.get(nd_url, params=params)
                                if r.status_code == 200:
                                    options = r.text.split()
                                    for option in options:
                                        if option.startswith('--volume-driver='):
                                            configs[image_name]['volume_driver'] = option.split('=', 1)[
                                                1]
                                        elif option.startswith('--volume='):
                                            vol = option.split('=', 1)[
                                                1].split(':')
                                            if 'volumes' in configs[image_name]:
                                                # !! TODO handle if volumes is a list
                                                configs[image_name]['volumes'][vol[0]] = {'bind': vol[1],
                                                                                          'mode': vol[2]}
                                            else:
                                                configs[image_name]['volumes'] = {vol[0]:
                                                                                  {'bind': vol[1],
                                                                                   'mode': vol[2]}}
                                        elif option.startswith('--device='):
                                            dev = option.split('=', 1)[1]
                                            if 'devices' in configs[image_name]:
                                                configs[image_name]['devices'].append(dev +
                                                                                      ':' +
                                                                                      dev +
                                                                                      ':rwm')
                                            else:
                                                configs[image_name]['devices'] = [
                                                    dev + ':' + dev + ':rwm']
                                        else:
                                            # unable to parse option provided by
                                            # nvidia-docker-plugin
                                            pass
                            except Exception as e:  # pragma: no cover
                                failed_images.add(image_name)
                                status = (False, str(e))
                                print('Failure with nvidia-docker-plugin: ' +
                                      str(e))
                except Exception as e:   # pragma: no cover
                    failed_images.add(image_name)
                    status = (False, str(e))
                    print('Unable to process gpu options: ' + str(e))
            path_cmd[image_name] = path
            orig_path_d[image_name] = orig_path
            labels_d[image_name] = labels

        # TODO get syslog address rather than hardcode
        # TODO add group label
        # TODO get group and name for syslog tag
        log_config = {'type': 'syslog',
                      'config': {'syslog-address': 'tcp://0.0.0.0:514',
                                 'syslog-facility': 'daemon',
                                 'tag': '{{.Name}}'}}

        # setup gpu queue
        can_queue_gpu = True
        try:
            q = Queue(connection=Redis(host=r_host), default_timeout=86400)
        except Exception as e:  # pragma: no cover
            can_queue_gpu = False
            print('Unable to connect to redis: ' + str(e))

        # start containers
        for image in images:
            if image not in failed_images:
                orig_path = orig_path_d[image]
                labels = labels_d[image]
                configs[image]['remove'] = True
                name = image.replace('/', '-').replace(':', '-') + '_' + \
                    str(int(time.time()))+'_'+str(uuid.uuid4())[:4]

                if orig_path:
                    # replay_pcap is special so we can't bind it like normal
                    # since the plugin takes in an additional argument
                    dir_path = orig_path.rsplit('/', 1)[0]
                else:
                    dir_path = path.rsplit('/', 1)[0]
                volumes = {dir_path: {'bind': dir_path, 'mode': 'rw'}}

                if 'volumes' in configs[image]:
                    for volume in volumes:
                        configs[image]['volumes'][volume] = volumes[volume]
                else:
                    configs[image]['volumes'] = volumes

                command = path_cmd[image]
                if 'command' in configs[image]:
                    command = configs[image]['command'] + ' ' + command
                    del configs[image]['command']

                if 'vent.gpu' in labels and labels['vent.gpu'] == 'yes':
                    if can_queue_gpu:
                        # queue up containers requiring a gpu
                        q_str = json.dumps({'image': image,
                                            'command': command,
                                            'labels': labels,
                                            'detach': True,
                                            'name': name,
                                            'log_config': log_config,
                                            'configs': configs[image]})
                        q.enqueue('watch.gpu_queue', q_str, ttl=2592000)
                    else:
                        failed_images.add(image)
                else:
                    if 'gpu_options' in configs[image]:
                        del configs[image]['gpu_options']
                    d_client.containers.run(image=image,
                                            command=command,
                                            labels=labels,
                                            detach=True,
                                            name=name,
                                            log_config=log_config,
                                            **configs[image])
        if failed_images:
            status = (False, failed_images)
        else:
            status = (True, images)
    except Exception as e:  # pragma: no cover
        status = (False, str(e))
        print('Error on line {}'.format(sys.exc_info()[-1].tb_lineno))
        print('Failed to process job: ' + str(e))

    print(str(status))
    return status
722442
private void process( Collection<File> templateFiles ) {

		// Compile them all
		Collection<TemplateEntry> templateEntries = new ArrayList<> ();
		for( File f : templateFiles ) {
			final TemplateEntry templateEntry = compileTemplate( f );
			if( templateEntry != null )
				templateEntries.add( templateEntry );
		}

		// Add all the template entries
		this.lock.writeLock().lock();
		try {
			for( final TemplateEntry te : templateEntries )
				this.fileToTemplate.put( te.getTemplateFile(), te );

		} finally {
			this.lock.writeLock().unlock();
		}

		// Notify the templating manager
		this.manager.processNewTemplates( templateEntries );
	}
935594
function process (filePath, complete) {
        var filename = path.basename(filePath),
            pathParts = path.dirname(filePath).split(path.sep)

        // we'll get an empty arrary if we overslice
        pathParts = pathParts.slice(transform.by)
        pathParts.push(filename)

        var newPath = pathParts.join(path.sep),
            fileData = files[filePath]

        // don't need the old fileData anymore,
        // since we are re-adding with the new path
        delete files[filePath]
        files[newPath] = fileData

        complete()
      }
49246
private void watchFilesTask() {
        while (!watcherExecutor.isShutdown()) {
            try {
                WatchKey key = watchService.take();
                Path parent = watchKeyDirs.get(key);
                if (parent == null) {
                    key.reset();
                    continue;
                }

                Set<Path> updates = new TreeSet<>();
                for (WatchEvent<?> ev : key.pollEvents()) {
                    WatchEvent.Kind<?> kind = ev.kind();

                    // Only file modification, creation, deletion is interesting.
                    if (kind != ENTRY_MODIFY &&
                        kind != ENTRY_CREATE &&
                        kind != ENTRY_DELETE) {
                        if (kind == OVERFLOW) {
                            LOGGER.warn("Overflow event, file updates may have been lost");
                        }
                        continue;
                    }

                    @SuppressWarnings("unchecked")
                    WatchEvent<Path> event = (WatchEvent<Path>) ev;

                    Path file = parent.resolve(event.context());
                    if (targetToRequests.containsKey(file)) {
                        LOGGER.trace("Watched file " + file + " event " + kind);
                        updates.add(file);
                    }
                }
                // Ready the key again so it wil signal more events.
                key.reset();

                if (updates.size() > 0) {
                    List<Supplier<Listener>>           watcherList;
                    Map<Path,List<Supplier<Listener>>> watcherMap;
                    Set<Path>                          updatedRequests = new TreeSet<>();
                    synchronized (mutex) {
                        watcherList = new ArrayList<>(this.watchers);
                        watcherMap = deepCopy(watchedFiles);

                        for (final Path file : updates) {
                            updatedRequests.addAll(targetToRequests.getOrDefault(file, Collections.emptySet()));
                        }
                        for (final Path file : updates) {
                            try {
                                updateLinkTargetting(file);
                            } catch (IOException e) {
                                LOGGER.warn("Failed to update link targeting: {}", e.getMessage(), e);
                            }
                        }
                    }

                    callbackExecutor.submit(() -> {
                        for (final Path file : updates) {
                            for (final Supplier<Listener> supplier : watcherList) {
                                Listener watcher = supplier.get();
                                if (watcher != null) {
                                    try {
                                        watcher.onPathUpdate(file);
                                    } catch (RuntimeException e) {
                                        LOGGER.error("Exception when notifying update on " + file, e);
                                    }
                                }
                            }
                        }
                        for (Path request : updatedRequests) {
                            Optional.ofNullable(watcherMap.get(request))
                                    .ifPresent(list -> {
                                        for (final Supplier<Listener> supplier : list) {
                                            Listener watcher = supplier.get();
                                            if (watcher != null) {
                                                try {
                                                    watcher.onPathUpdate(request);
                                                } catch (RuntimeException e) {
                                                    LOGGER.error("Exception when notifying update on " + request, e);
                                                }
                                            }
                                        }
                                    });
                        }
                    });
                }
            } catch (InterruptedException interruptedEx) {
                LOGGER.error("Interrupted in service file watch thread: " + interruptedEx.getMessage(), interruptedEx);
                return;
            }
        }
    }
1126590
def process_files
      if @options[:parallel].empty?
        files.map { |path| check_path(path) }
      else
        Parallel.map(files, @options[:parallel]) { |path| check_path(path) }
      end
    end
623828
public function processCombinedFiles()
    {
        // Check if combining is enabled
        if (!$this->getCombinedFilesEnabled()) {
            return;
        }

        // Before scripts are modified, detect files that are provided by preceding ones
        $providedScripts = $this->getProvidedScripts();

        // Process each combined files
        foreach ($this->getAllCombinedFiles() as $combinedFile => $combinedItem) {
            $fileList = $combinedItem['files'];
            $type = $combinedItem['type'];
            $options = $combinedItem['options'];

            // Generate this file, unless blocked
            $combinedURL = null;
            if (!isset($this->blocked[$combinedFile])) {
                // Filter files for blocked / provided
                $filteredFileList = array_diff(
                    $fileList,
                    $this->getBlocked(),
                    $providedScripts
                );
                $combinedURL = $this->getCombinedFileURL($combinedFile, $filteredFileList, $type);
            }

            // Replace all existing files, injecting the combined file at the position of the first item
            // in order to preserve inclusion order.
            // Note that we iterate across blocked files in order to get the correct order, and validate
            // that the file is included in the correct location (regardless of which files are blocked).
            $included = false;
            switch ($type) {
                case 'css': {
                    $newCSS = array(); // Assoc array of css file => spec
                    foreach ($this->getAllCSS() as $css => $spec) {
                        if (!in_array($css, $fileList)) {
                            $newCSS[$css] = $spec;
                        } elseif (!$included && $combinedURL) {
                            $newCSS[$combinedURL] = array('media' => (isset($options['media']) ? $options['media'] : null));
                            $included = true;
                        }
                        // If already included, or otherwise blocked, then don't add into CSS
                    }
                    $this->css = $newCSS;
                    break;
                }
                case 'js': {
                    // Assoc array of file => attributes
                    $newJS = array();
                    foreach ($this->getAllJavascript() as $script => $attributes) {
                        if (!in_array($script, $fileList)) {
                            $newJS[$script] = $attributes;
                        } elseif (!$included && $combinedURL) {
                            $newJS[$combinedURL] = $options;
                            $included = true;
                        }
                        // If already included, or otherwise blocked, then don't add into scripts
                    }
                    $this->javascript = $newJS;
                    break;
                }
            }
        }
    }
150489
public function process()
    {

        if ($this->plugin->request()->has('_flush')) {

            if ($this->flushCache($this->plugin->request()->get('_flush'))) {

                $this->attributes['alertClass'] = 'success';
                $this->attributes['messages'] = array(
                    '&#10004; ' . $this->plugin->config()->get('plugin.namespace') . ' ' . ucwords($this->plugin->request()->get('_flush')) . ' flushed successfully.'
                );
                //Refresh Data
            } else {

                $this->attributes['alertClass'] = 'error';
                $this->attributes['messages'] = array(
                    '&#9888; ' . $this->plugin->config()->get('plugin.namespace') . ' ' . ucwords($this->plugin->request()->get('_flush'))
                    .' could not be flushed.  The file(s) may not exist; If they exist, check directory structure, permissions and config paths.'
                );
            }

            $this->data();
        }

    }
607311
def process_files():
    

    with enlighten.Manager() as manager:
        with manager.counter(total=SPLINES, desc='Reticulating:', unit='splines') as retic:
            for num in range(SPLINES):  # pylint: disable=unused-variable
                time.sleep(random.uniform(0.1, 0.5))  # Random processing time
                retic.update()

        with manager.counter(total=LLAMAS, desc='Herding:', unit='llamas') as herd:
            for num in range(SPLINES):  # pylint: disable=unused-variable
                time.sleep(random.uniform(0.1, 0.5))  # Random processing time
                herd.update()
404488
private void processFiles() {
        String[] fileArgs = getOptionValues(INFILE_SHORT_OPT);
        final List<File> files = new ArrayList<File>(fileArgs.length);
        for (final String fileArg : fileArgs) {
            File file = new File(fileArg);
            if (!file.canRead()) {
                error(INPUT_FILE_UNREADABLE + file);
            } else {
                files.add(file);
            }
        }

        if (files.size() == 0) {
            error(NO_DOCUMENT_FILES);
            failUsage();
        }

        // Create the directory artifact or fail
        artifactPath = createDirectoryArtifact(outputDirectory, DIR_ARTIFACT);
        processFiles(files.toArray(new File[0]));
    }
711665
private function processingSourceFiles(): void
  {
    $this->logVerbose('Replace references to resource files with references to optimized/minimized resource files.');

    foreach ($this->sourceFilesInfo as $source_filename)
    {
      $this->logVerbose("Processing %s.", $source_filename);

      $content = file_get_contents($source_filename);
      if ($content===false) $this->logError("Unable to read file '%s'.", $source_filename);

      $new_content = $content;

      if (strncmp($content, '<?php', 5)===0)
      {
        // Source is a PHP file.
        $new_content = $this->processPhpSourceFile($source_filename, $new_content);
      }

      $new_content = strtr($new_content, $this->replacePairs);

      if ($content!=$new_content)
      {
        $time = null;

        // If required determine the latest modification time of the source file and its referenced resource files.
        if ($this->preserveModificationTime)
        {
          $time = $this->getMaxModificationTime($source_filename, $new_content);
        }

        // Write sources file with modified references to resource files.
        $status = file_put_contents($source_filename, $new_content);
        if ($status===false) $this->logError("Updating file '%s' failed.", $source_filename);
        $this->logInfo("Updated file '%s'.", $source_filename);

        // If required set the mtime to the latest modification time of the source file and its referenced resource
        // files.
        if ($this->preserveModificationTime)
        {
          $status = touch($source_filename, $time);
          if ($status===false) $this->logError("Unable to set mtime for file '%s'.", $source_filename);
        }
      }
    }
  }
875987
def process_file(filename, source):
    ''''''
    print("Processing %s" % filename)
    mlog = mavutil.mavlink_connection(filename, notimestamps=args.notimestamps)
    
    position_field_type = sniff_field_spelling(mlog, source)
    
    # init fields and field_types lists
    fields = [args.source + "." + s for s in position_field_type]
    fields.append(mainstate_field)
    field_types = []

    msg_types = set()
    re_caps = re.compile('[A-Z_][A-Z0-9_]+')

    for f in fields:
        caps = set(re.findall(re_caps, f))
        msg_types = msg_types.union(caps)
        field_types.append(caps)
    
    add_data.new_linestring = True
    add_data.mainstate_current = -1
    add_data.current_kml_linestring = None
    add_data.position_data = [None for n in position_field_type]
    add_data.last_time = 0

    while True:
        msg = mlog.recv_match(args.condition)
        if msg is None:
            break
        tdays = (msg._timestamp - time.timezone) / (24 * 60 * 60)
        tdays += 719163  # pylab wants it since 0001-01-01
        add_data(tdays, msg, msg_types, mlog.messages, fields, field_types, position_field_type)
317720
private void processFiles() throws CompilationFailedException, IOException, URISyntaxException {
        GroovyShell groovy = new GroovyShell(conf);
        setupContextClassLoader(groovy);

        Script s = groovy.parse(getScriptSource(isScriptFile, script));

        if (args.isEmpty()) {
            try(BufferedReader reader = new BufferedReader(new InputStreamReader(System.in))) {
                PrintWriter writer = new PrintWriter(System.out);
                processReader(s, reader, writer);
                writer.flush();
            }
        } else {
            Iterator i = args.iterator();
            while (i.hasNext()) {
                String filename = (String) i.next();
                //TODO: These are the arguments for -p and -i.  Why are we searching using Groovy script extensions?
                // Where is this documented?
                File file = huntForTheScriptFile(filename);
                processFile(s, file);
            }
        }
    }
462738
def _postprocess(self, filehandle, metadata):
        ""
        for process in self._postprocessors:
            filehandle = process(filehandle, metadata)
        return filehandle
639717
function processFile() {
  return (file, callback) => {
    let parsed;
    if (file.meta !== undefined && file.meta.layout !== undefined) {
      const ext = path.extname(file.meta.layout);
      parsed = path.parse(`${trimExtension(file.path)}${ext}`);
    } else {
      parsed = path.parse(file.path);
    }

    const filePath = {
      dir: parsed.dir,
      name: parsed.name,
      ext: parsed.ext,
    };
    const meta = Object.assign({}, file.meta, { filePath });
    async.nextTick(callback, null, Object.assign({}, file, { meta }));
  };
}
200281
def _ProcessRepliesWithFlowOutputPlugins(self, replies):
    """"""
    created_output_plugins = []
    for index, output_plugin_state in enumerate(
        self.rdf_flow.output_plugins_states):
      plugin_descriptor = output_plugin_state.plugin_descriptor
      output_plugin_cls = plugin_descriptor.GetPluginClass()
      output_plugin = output_plugin_cls(
          source_urn=self.rdf_flow.long_flow_id,
          args=plugin_descriptor.plugin_args,
          token=access_control.ACLToken(username=self.rdf_flow.creator))

      try:
        # TODO(user): refactor output plugins to use FlowResponse
        # instead of GrrMessage.
        output_plugin.ProcessResponses(
            output_plugin_state.plugin_state,
            [r.AsLegacyGrrMessage() for r in replies])
        output_plugin.Flush(output_plugin_state.plugin_state)
        output_plugin.UpdateState(output_plugin_state.plugin_state)

        data_store.REL_DB.WriteFlowOutputPluginLogEntries([
            rdf_flow_objects.FlowOutputPluginLogEntry(
                client_id=self.rdf_flow.client_id,
                flow_id=self.rdf_flow.flow_id,
                hunt_id=self.rdf_flow.parent_hunt_id,
                output_plugin_id="%d" % index,
                log_entry_type=rdf_flow_objects.FlowOutputPluginLogEntry
                .LogEntryType.LOG,
                message="Processed %d replies." % len(replies))
        ])

        self.Log("Plugin %s successfully processed %d flow replies.",
                 plugin_descriptor, len(replies))

        created_output_plugins.append(output_plugin)
      except Exception as e:  # pylint: disable=broad-except
        logging.exception("Plugin %s failed to process %d replies: %s",
                          plugin_descriptor, len(replies),
                          traceback.format_exc())
        created_output_plugins.append(None)

        data_store.REL_DB.WriteFlowOutputPluginLogEntries([
            rdf_flow_objects.FlowOutputPluginLogEntry(
                client_id=self.rdf_flow.client_id,
                flow_id=self.rdf_flow.flow_id,
                hunt_id=self.rdf_flow.parent_hunt_id,
                output_plugin_id="%d" % index,
                log_entry_type=rdf_flow_objects.FlowOutputPluginLogEntry
                .LogEntryType.ERROR,
                message="Error while processing %d replies: %s" %
                (len(replies), str(e)))
        ])

        self.Log("Plugin %s failed to process %d replies due to: %s",
                 plugin_descriptor, len(replies), e)

    return created_output_plugins
599628
def process_files(self):
        
        from .files import VideoFile, ExtractedVideoThumbnailFile, WebVideoFile

        downloaded = super(VideoNode, self).process_files()

        try:
            # Extract thumbnail if one hasn't been provided and derive_thumbnail is set
            if self.generate_thumbnail and not self.has_thumbnail():
                videos = [f for f in self.files if isinstance(f, VideoFile) or isinstance(f, WebVideoFile)]
                assert len(videos) > 0 and videos[0].filename, "Cannot extract thumbnail (No videos found on node {0})".format(self.source_id)

                self.set_thumbnail(ExtractedVideoThumbnailFile(config.get_storage_path(videos[0].filename)))
                downloaded.append(self.thumbnail.get_filename())

        except AssertionError as ae:
            config.LOGGER.warning(ae)

        return downloaded
1287221
function processFile(sourcePath, templates, customPartialName, files) {
  return (file, callback) => {
    const layout = templates.layouts[file.meta.layout];
    const fns = wrapFunctions(templates.fns, file, files);
    const context = Object.assign({}, file.meta, fns);
    async.waterfall([
      cb => getPartials(file, sourcePath, templates, customPartialName, cb),
      (partials, cb) => render(file, layout, context, partials, cb),
    ], (error, result) => {
      const data = result;
      async.nextTick(callback, error, Object.assign({}, file, { data }));
    });
  };
}