Opens a file for writing using the properties of the event. All data from
subsequent {@link Input} events is written to the file.
The end of record flag is ignored.

@param event the event
@throws InterruptedException if the execution was interrupted
358880
@Handler
    @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
    public void onSaveInput(SaveInput event) throws InterruptedException {
        if (!Arrays.asList(event.options())
            .contains(StandardOpenOption.WRITE)) {
            throw new IllegalArgumentException(
                "File must be opened for writing.");
        }
        for (IOSubchannel channel : event.channels(IOSubchannel.class)) {
            if (inputWriters.containsKey(channel)) {
                channel.respond(new IOError(event,
                    new IllegalStateException("File is already open.")));
            } else {
                new Writer(event, channel);
            }
        }
    }
968673
def WriteEvent(self, event):
    
    self.WriteEventStart()

    try:
      self.WriteEventBody(event)

    except errors.NoFormatterFound as exception:
      error_message = 'unable to retrieve formatter with error: {0!s}'.format(
          exception)
      self._ReportEventError(event, error_message)

    except errors.WrongFormatter as exception:
      error_message = 'wrong formatter with error: {0!s}'.format(exception)
      self._ReportEventError(event, error_message)

    self.WriteEventEnd()
686021
public void copyFrom(InputStream in) throws IOException, InterruptedException {
        try (OutputStream os = write()) {
            org.apache.commons.io.IOUtils.copy(in, os);
        }
    }
1221521
private Event createEvent(String obs) {
        int lastSpace = obs.lastIndexOf(StringUtils.SPACE);
        Event event = null;

        if (lastSpace != -1) {
            String label = obs.substring(lastSpace + 1);
            String[] contexts = obs.substring(0, lastSpace).split("\\s+");
            // Split name and value
            float[] values = RealValueFileEventStream.parseContexts(contexts);
            // Label, feature name, feature value
            event = new Event(label, contexts, values);
        }

        return event;
    }
1096264
@Override
    public void uploadStarted(final StartedEvent event) {
        // reset internal state here because instance is reused for next upload!
        resetState();

        final SoftwareModule softwareModule = getSelectedSoftwareModule();

        this.fileUploadId = new FileUploadId(event.getFilename(), softwareModule);

        if (getUploadState().isFileInUploadState(this.fileUploadId)) {
            // actual interrupt will happen a bit late so setting the below
            // flag
            interruptUploadDueToDuplicateFile();
            event.getUpload().interruptUpload();
        } else {
            LOG.info("Uploading file {}", fileUploadId);
            publishUploadStarted(fileUploadId);

            if (RegexCharacterCollection.stringContainsCharacter(event.getFilename(), ILLEGAL_FILENAME_CHARACTERS)) {
                LOG.info("Filename contains illegal characters {} for upload {}", fileUploadId.getFilename(),
                        fileUploadId);
                interruptUploadDueToIllegalFilename();
                event.getUpload().interruptUpload();
            } else if (isFileAlreadyContainedInSoftwareModule(fileUploadId, softwareModule)) {
                LOG.info("File {} already contained in Software Module {}", fileUploadId.getFilename(), softwareModule);
                interruptUploadDueToDuplicateFile();
                event.getUpload().interruptUpload();
            }
        }
    }
563327
@Override
	public void reopen(FileInputSplit split, Tuple2<Long, Long> state) throws IOException {
		this.open(split);
		this.getOfficeReader().getCurrentParser().setCurrentSheet(state.f0);
		this.getOfficeReader().getCurrentParser().setCurrentRow(state.f1);

	}
1280472
public synchronized void insert_event(EventData event) {
        events.add(event);
        while (events.size() > max_size)
            events.remove(0);
    }
342469
public void sendEvent(Event event) {
        for (ShouldSendEventCallback shouldSendEventCallback : shouldSendEventCallbacks) {
            if (!shouldSendEventCallback.shouldSend(event)) {
                logger.trace("Not sending Event because of ShouldSendEventCallback: {}", shouldSendEventCallback);
                return;
            }
        }

        try {
            connection.send(event);
        } catch (LockedDownException | TooManyRequestsException e) {
            logger.debug("Dropping an Event due to lockdown: " + event);
        } catch (Exception e) {
            logger.error("An exception occurred while sending the event to Sentry.", e);
        } finally {
            getContext().setLastEventId(event.getId());
        }
    }
1340138
@Handler
    public void onInput(Input<ByteBuffer> event, IOSubchannel channel)
            throws InterruptedException, UnsupportedEncodingException {
        Optional<FormContext> ctx = channel.associated(this, FormContext.class);
        if (!ctx.isPresent()) {
            return;
        }
        ctx.get().fieldDecoder.addData(event.data());
        if (!event.isEndOfRecord()) {
            return;
        }

        long invocations = (Long) ctx.get().session.computeIfAbsent(
            "invocations", key -> {
                return 0L;
            });
        ctx.get().session.put("invocations", invocations + 1);

        HttpResponse response = ctx.get().request.response().get();
        response.setStatus(HttpStatus.OK);
        response.setHasPayload(true);
        response.setField(HttpField.CONTENT_TYPE,
            MediaType.builder().setType("text", "plain")
                .setParameter("charset", "utf-8").build());
        String data = "First name: "
            + ctx.get().fieldDecoder.fields().get("firstname")
            + "\r\n" + "Last name: "
            + ctx.get().fieldDecoder.fields().get("lastname")
            + "\r\n" + "Previous invocations: " + invocations;
        channel.respond(new Response(response));
        ManagedBuffer<ByteBuffer> out = channel.byteBufferPool().acquire();
        out.backingBuffer().put(data.getBytes("utf-8"));
        channel.respond(Output.fromSink(out, true));
    }
1317551
private void processEvent(EventModel<?> event) {
        if (!event.getSource().isCreatedFromInstance()) {
            error("event: " + event + "has invalid source");
            return;
        }
        debug("EventFired: " + event.toString() + " from " + event.getSource().getID());
        submit(() -> event.lifecycleCallback(EventLifeCycle.START));

        if (checkEventsControllers(event)) {
            submit(() -> event.lifecycleCallback(EventLifeCycle.APPROVED));
            submit(() -> event.lifecycleCallback(EventLifeCycle.RESOURCE));
            List<ResourceModel> resourceList = getMain().getResourceManager().generateResources(event);
            event.addResources(resourceList);
            submit(() -> event.lifecycleCallback(EventLifeCycle.LISTENERS));
            List<EventListenerModel> listenersTemp = event.getAllInformations().parallelStream()
                    .map(listeners::get)
                    .filter(Objects::nonNull)
                    .flatMap(Collection::stream)
                    .distinct()
                    .collect(Collectors.toList());

            List<CompletableFuture> futures = listenersTemp.stream()
                    .map(eventListener -> submit(() -> eventListener.eventFired(event)))
                    .collect(Collectors.toList());
            try {
                timeOut(futures, 1000);
            } catch (InterruptedException e) {
                error("interrupted", e);
            }
            submit(() -> event.lifecycleCallback(EventLifeCycle.OUTPUT));
            getMain().getOutputManager().passDataToOutputPlugins(event);
            submit(() -> event.lifecycleCallback(EventLifeCycle.ENDED));
            List<EventListenerModel> finishListenersTemp = event.getAllInformations().parallelStream()
                    .map(finishListeners::get)
                    .filter(Objects::nonNull)
                    .flatMap(Collection::stream)
                    .distinct()
                    .collect(Collectors.toList());

            futures = finishListenersTemp.stream()
                    .map(eventListener -> submit(() -> eventListener.eventFired(event)))
                    .collect(Collectors.toList());

            try {
                timeOut(futures, 1000);
            } catch (InterruptedException e) {
                error("interrupted", e);
            }
        } else {
            debug("canceling: " + event.toString() + " from " + event.getSource().getID());
            submit(() -> event.lifecycleCallback(EventLifeCycle.CANCELED));
        }
    }
593401
public void reopen(Iterator<T> input) throws IOException {
		this.input = input;
		
		this.noMoreBlocks = false;
		this.closed = false;
		
		nextBlock();
	}
815161
@Override
    public void add(Event event) {
        if (getNumStoredEvents() >= maxEvents) {
            logger.warn("Not adding Event because at least "
                + Integer.toString(maxEvents) + " events are already stored: " + event.getId());
            return;
        }

        File eventFile = new File(bufferDir.getAbsolutePath(), event.getId().toString() + FILE_SUFFIX);
        if (eventFile.exists()) {
            logger.trace("Not adding Event to offline storage because it already exists: "
                + eventFile.getAbsolutePath());
            return;
        } else {
            logger.debug("Adding Event to offline storage: " + eventFile.getAbsolutePath());
        }

        try (FileOutputStream fileOutputStream = new FileOutputStream(eventFile);
             ObjectOutputStream objectOutputStream = new ObjectOutputStream(fileOutputStream)) {
            objectOutputStream.writeObject(event);
        } catch (Exception e) {
            logger.error("Error writing Event to offline storage: " + event.getId(), e);
        }

        logger.debug(Integer.toString(getNumStoredEvents())
            + " stored events are now in dir: "
            + bufferDir.getAbsolutePath());
    }
1190784
public void doAppend(LoggingEvent event)
    {
        if(closed) {
            LogLog.error("Attempted to append to closed appender named [" + name + "].");
            return;
        }

        if(!isAsSevereAsThreshold(event.getLevel())) {
            return;
        }

        Filter f = this.headFilter;

        FILTER_LOOP:
        while(f != null) {
            switch(f.decide(event)) {
            case Filter.DENY: return;
            case Filter.ACCEPT: break FILTER_LOOP;
            case Filter.NEUTRAL: f = f.getNext();
            }
        }

        this.append(event);
    }
212241
@Override
	public void open(FileInputSplit split) throws IOException {
		super.open(split);

		// instantiate the parsers
		FieldParser<?>[] parsers = new FieldParser<?>[fieldTypes.length];
		
		for (int i = 0; i < fieldTypes.length; i++) {
			if (fieldTypes[i] != null) {
				Class<? extends FieldParser<?>> parserType = FieldParser.getParserForType(fieldTypes[i]);
				if (parserType == null) {
					throw new RuntimeException("No parser available for type '" + fieldTypes[i].getName() + "'.");
				}

				FieldParser<?> p = InstantiationUtil.instantiate(parserType, FieldParser.class);

				p.setCharset(getCharset());
				if (this.quotedStringParsing) {
					if (p instanceof StringParser) {
						((StringParser)p).enableQuotedStringParsing(this.quoteCharacter);
					} else if (p instanceof StringValueParser) {
						((StringValueParser)p).enableQuotedStringParsing(this.quoteCharacter);
					}
				}

				parsers[i] = p;
			}
		}
		this.fieldParsers = parsers;
		
		// skip the first line, if we are at the beginning of a file and have the option set
		if (this.skipFirstLineAsHeader && this.splitStart == 0) {
			readLine(); // read and ignore
		}
	}
99766
private void init()
        throws IOException, InterruptedException {
        nakedDone = false;

        // initialize the total length
        float recToFragRatio = conf.getFloat(RECORD_TO_FRAGMENT_RATIO,
            getDefaultRatio());
        length = mlSplit.getLength() * recToFragRatio;

        // generate the query
        String queryText;
        long start = mlSplit.getStart() + 1 + count;
        long end = mlSplit.isLastSplit() ? Long.MAX_VALUE : start
            + mlSplit.getLength() - count - 1;

        String src = conf.get(MarkLogicConstants.DOCUMENT_SELECTOR);
        redactionRuleCol = conf.getStrings(REDACTION_RULE_COLLECTION);
        Collection<String> nsCol = null;
        if (src != null) {
            nsCol = conf.getStringCollection(
                    MarkLogicConstants.PATH_NAMESPACE);
        } else {
            src = "fn:collection()";
        }
        ctsQuery = conf.get(MarkLogicConstants.QUERY_FILTER);
        StringBuilder buf = new StringBuilder();
        if (ctsQuery != null) {
            buildSearchQuery(src, ctsQuery, nsCol, buf);
        } else {
            buildDocExprQuery(src, nsCol, null, buf);
        }   
        src = buf.toString();
        buf = new StringBuilder();
        buf.append("xquery version \"1.0-ml\"; \n");
        buf.append("import module namespace hadoop = ");
        buf.append("\"http://marklogic.com/xdmp/hadoop\" at ");
        buf.append("\"/MarkLogic/hadoop.xqy\";\n");
        if (redactionRuleCol != null) {
            buf.append(
                "import module namespace rdt = \"http://marklogic.com/xdmp/redaction\" at \"/MarkLogic/redaction.xqy\";\n");
        }
        buf.append(
                "declare namespace mlmr=\"http://marklogic.com/hadoop\";\n");
        buf.append("declare option xdmp:output \"indent=no\";\n");
        buf.append("declare option xdmp:output \"indent-untyped=no\";\n");
        buf.append(
                "declare variable $mlmr:splitstart as xs:integer external;\n");
        buf.append(
                "declare variable $mlmr:splitend as xs:integer external;\n");
        buf.append("let $cols := ");
        buf.append(src);
        buf.append("\nlet $all-meta :=");
        buf.append("\nfor $doc in $cols");
        buf.append("\nlet $uri := fn:base-uri($doc)\n return (");

        buf.append("'META',");
        buf.append("$uri,");
        buf.append("if(fn:empty($doc/node())) then 0 ");
        buf.append("else if (fn:count($doc/node())>1) then \"element\" ");
        buf.append("else xdmp:node-kind($doc/node())");
        if (copyCollection || copyPermission || copyProperties || copyQuality) {
            buf.append(",");
            if (copyCollection) {
                buf.append("xdmp:document-get-collections($uri),\n");
            }
            if (copyPermission) {
                buf.append("let $list := xdmp:document-get-permissions($uri)\n");
                buf.append("return hadoop:get-permissions($list),");
            }
            // if copy-quality, else + 0
            if (copyQuality) {
                buf.append("xdmp:document-get-quality($uri),\n");
            } else {
                buf.append("0,");
            }
            // if copy-metadata
            if (copyMetadata) {
                buf.append(
                    "(let $f := fn:function-lookup(xs:QName('xdmp:document-get-metadata'),1)\n"
                    + "return if (exists($f)) then $f($uri) else ()),\n");
            }
            // if copy-properties, else + (),\n
            if (copyProperties) {
                buf.append("xdmp:document-properties($uri)/prop:properties,\n");
            }
        } else {
            buf.append(",0,"); // quality
            buf.append("(),\n");//properties
        }
        // end-of-record marker
        buf.append("0");       
        buf.append(" )\n");
        buf.append("return ($all-meta");
        buf.append(",'EOM',$cols)");

        queryText = buf.toString();
        if (LOG.isDebugEnabled()) {
            LOG.debug(queryText);
        }

        // set up a connection to the server
        while (retry < maxRetries) {
        try {
            if (retry == 1) {
                LOG.info("Retrying connect");
            }
            String curForestName = "";
            String curHostName = "";
            if (curForest == -1) {
                curForestName = mlSplit.getForestId().toString();
                curHostName = hostNames[0];
            } else {
                curForestName = replicas.get(curForest).getForest();
                curHostName = replicas.get(curForest).getHostName();
            }
            ContentSource cs = InternalUtilities.getInputContentSource(conf,
                    curHostName);
            session = cs.newSession("#"+curForestName);
            if (LOG.isDebugEnabled()) {
                LOG.debug("Connect to forest "
                    + curForestName + " on "
                    + session.getConnectionUri().getHost());
            }

            AdhocQuery aquery = session.newAdhocQuery(queryText);
            aquery.setNewIntegerVariable(MR_NAMESPACE, SPLIT_START_VARNAME,
                start);
            aquery.setNewIntegerVariable(MR_NAMESPACE, SPLIT_END_VARNAME, end);
            RequestOptions options = new RequestOptions();
            options.setCacheResult(false);
            String ts = conf.get(INPUT_QUERY_TIMESTAMP);
            if (ts != null) {
                options.setEffectivePointInTime(new BigInteger(ts));
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Query timestamp: " + ts);
                }
            } 
            aquery.setOptions(options);
            result = session.submitRequest(aquery);
            
            initMetadataMap();
        } catch (XccConfigException e) {
            LOG.error("XccConfigException:" + e);
            throw new IOException(e);
        } catch (QueryException e) {
            LOG.error("QueryException:" + e);
            LOG.debug("Query: " + queryText);
            throw new IOException(e);
        } catch (Exception e) {
            LOG.error("Exception:" + e.getMessage());
            if (curForest != -1) {
                if (++retry < maxRetries) {
                    // failover
                    try {
                        Thread.sleep(sleepTime);
                    } catch (Exception e2) {
                    }
                    sleepTime = Math.min(sleepTime * 2,maxSleepTime);

                    curForest = (curForest+1)%replicas.size();
                    continue;
                }
                LOG.info("Retry limit exceeded");
            }
            throw new IOException(e);
        }
        break;
        }
    }
1372976
public void postEvent (DEvent event)
    {
        if (!_running) {
            log.warning("Posting message to inactive object manager", "event", event,
                        new Exception());
        }

        // assign the event's id and append it to the queue
        event.eventId = getNextEventId(true);
        _evqueue.append(event);
    }
791299
private void batchFileButtonActionPerformed(java.awt.event.ActionEvent evt) {//GEN-FIRST:event_batchFileButtonActionPerformed
        // Load a batch file
        JFileChooser fileChooser = new JFileChooser();
        if (fileChooser.showOpenDialog(this) == JFileChooser.APPROVE_OPTION) {
            File file = fileChooser.getSelectedFile();
            try (BufferedReader in = new BufferedReader(
                    new InputStreamReader(
                        new FileInputStream(file), "UTF8"))) {
                String str;

                sequenceArea.setText("");

                while ((str = in.readLine()) != null) {
                    sequenceArea.setText(sequenceArea.getText() + str + '\n');
                }
            } catch (UnsupportedEncodingException e) {
                System.out.println("Encoding exception");
            } catch (IOException e) {
                System.out.println("Cannot read from file" + fileChooser.getSelectedFile().toString());
            } catch (Exception e) {
                System.out.println("Exception");
            }

        }
    }