Storm Pipe Network File Write to File Method
540934
def _write(self, session, openFile, replaceParamFile):
        
        # Retrieve Connection objects and write to file
        connections = self.connections
        self._writeConnections(connections=connections,
                               fileObject=openFile)

        # Retrieve SuperJunction objects and write to file
        sjuncs = self.superJunctions
        self._writeSuperJunctions(superJunctions=sjuncs,
                                  fileObject=openFile)

        # Retrieve SuperLink objects and write to file
        slinks = self.superLinks
        self._writeSuperLinks(superLinks=slinks,
                              fileObject=openFile)
407162
def write(self, node, filehandle):
        """"""
        dictexporter = self.dictexporter or DictExporter()
        data = dictexporter.export(node)
        return json.dump(data, filehandle, **self.kwargs)
577461
def _write(self, session, openFile, replaceParamFile):
        
        # Write lines
        openFile.write('GRIDSTREAMFILE\n')
        openFile.write('STREAMCELLS %s\n' % self.streamCells)

        for cell in self.gridStreamCells:
            openFile.write('CELLIJ    %s  %s\n' % (cell.cellI, cell.cellJ))
            openFile.write('NUMNODES  %s\n' % cell.numNodes)

            for node in cell.gridStreamNodes:
                openFile.write('LINKNODE  %s  %s  %.6f\n' % (
                    node.linkNumber,
                    node.nodeNumber,
                    node.nodePercentGrid))
47550
function processFile(rootPath, log) {
  return (file, callback) => {
    const filePath = path.join(rootPath, file.path);
    const writeMode = isStream(file.data) ? WriteMode.STREAM : WriteMode.UTF8;
    write(filePath, file.data, writeMode, (error) => {
      if (log === true && !error) {
        console.log(`Successfully wrote file '${file.path}'`);
      }

      async.nextTick(callback, error, file);
    });
  };
}
263946
@Override
    public long write(long numBytes, int time) throws IOException {
        int timeout = time;
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "write(" + numBytes + "," + timeout + ")");
        }
        getTCPConnLink().incrementNumWrites();
        if (getConfig().getDumpStatsInterval() > 0) {
            getTCPConnLink().getTCPChannel().totalSyncWrites.incrementAndGet();
        }
        // always check external call parms
        checkForErrors(numBytes, false, timeout);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
            Socket socket = getTCPConnLink().getSocketIOChannel().getSocket();
            Tr.event(tc, "write (sync) requested for local: " + socket.getLocalSocketAddress() + " remote: " + socket.getRemoteSocketAddress());
        }

        long numUserBytesWritten = 0L; // no. of user bytes written

        if (isAborted()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Previously aborted, unable to perform write");
            }
            throw new IOException("Connection aborted by program");
        }

        if (timeout == IMMED_TIMEOUT) {
            immediateTimeout();
        } else if (timeout == ABORT_TIMEOUT) {
            abort();
            immediateTimeout();
        } else {
            // if using channel timeout, reset to that value
            if (timeout == TCPRequestContext.USE_CHANNEL_TIMEOUT) {
                timeout = getConfig().getInactivityTimeout();
            }
            // check if this is a FileChannel write request
            if (isFileChannelWriteToBeDone()) {
                numUserBytesWritten = fileChannelWrite(numBytes, timeout);
            } else {
                numUserBytesWritten = processSyncWriteRequest(numBytes, timeout);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "write(sync): " + numUserBytesWritten);
        }
        return (numUserBytesWritten);
    }
69126
def write_file(path)
      exec "mkdir -p #{File.dirname(path).gsub(/ /, "\\ ")}"
      Tempfile.open(File.basename(path)) do |f|
        yield f
        f.flush
        remote_path = %Q{"#{@user}@#{host}:#{path.gsub(/ /, "\\ ")}"}
        puts "Writing to #{remote_path}"
        `scp '#{f.path}' #{remote_path}`
      end
    end
707270
def write(self, chunk):
        """"""
        if self.chunked_write and chunk:
            buf = [hex(len(chunk))[2:], CRLF, chunk, CRLF]
            self.conn.wfile.sendall("".join(buf))
        else:
            self.conn.wfile.sendall(chunk)
782249
private static void writeFile(CyclicBarrier barrier, AtomicLong runtime, int count,
      AlluxioConfiguration alluxioConf)
      throws Exception {
    FileSystem fileSystem = FileSystem.Factory.create(alluxioConf);
    byte[] buffer = new byte[(int) Math.min(sFileSize, 4 * Constants.MB)];
    Arrays.fill(buffer, (byte) 'a');
    AlluxioURI path = filename(count);

    if (fileSystem.exists(path)) {
      fileSystem.delete(path);
    }

    barrier.await();
    long startTime = System.nanoTime();
    long bytesWritten = 0;
    try (FileOutStream outStream = fileSystem.createFile(path)) {
      while (bytesWritten < sFileSize) {
        outStream.write(buffer, 0, (int) Math.min(buffer.length, sFileSize - bytesWritten));
        bytesWritten += buffer.length;
      }
    }
    runtime.addAndGet(System.nanoTime() - startTime);
  }
1038899
public void launchWorker(Map conf, IContext sharedContext, String topologyId, String supervisorId,
                             Integer port, String workerId, LocalAssignment assignment) throws IOException {
        // get supervisor conf
        Map stormConf = StormConfig.read_supervisor_topology_conf(conf, topologyId);
        String stormHome = System.getProperty("jstorm.home");
        if (StringUtils.isBlank(stormHome)) {
            stormHome = "./";
        }

        // get worker conf
        String topologyRoot = StormConfig.supervisor_stormdist_root(conf, topologyId);
        Map workerConf = StormConfig.read_topology_conf(topologyRoot, topologyId);

        Map totalConf = new HashMap();
        totalConf.putAll(conf);
        totalConf.putAll(stormConf);
        totalConf.putAll(workerConf);

        /**
         * don't use environment.putAll(System.getenv);
         */
        Map<String, String> environment = new HashMap<>();
        if (ConfigExtension.getWorkerRedirectOutput(totalConf)) {
            environment.put("REDIRECT", "true");
        } else {
            environment.put("REDIRECT", "false");
        }
        environment.put("LD_LIBRARY_PATH", (String) totalConf.get(Config.JAVA_LIBRARY_PATH));
        environment.put("jstorm.home", stormHome);
        environment.put("jstorm.workerId", workerId);
        environment.put("jstorm.on.yarn", isJstormOnYarn ? "1" : "0");

        String launcherCmd = getLauncherParameter(assignment, totalConf, stormHome, topologyId, port);
        String workerCmd = getWorkerParameter(assignment,
                totalConf,
                stormHome,
                topologyId,
                supervisorId,
                workerId,
                port);
        String cmd = launcherCmd + " " + workerCmd;
        cmd = cmd.replace("%JSTORM_HOME%", stormHome);

        LOG.info("Launching worker with command: " + cmd);
        LOG.info("Environment:" + environment.toString());

        /**
         * if run on yarn, set backend false, otherwise set true
         */
        boolean backend = !isJstormOnYarn;
        LOG.info("backend mode is " + backend);
        JStormUtils.launchProcess(cmd, environment, backend);
    }
900911
@Nonnull
    private static File writeDataToTempFileOrDie(@Nonnull final OutputStreamCallback callback,
                                                 @Nonnull final File targetFile,
                                                 @Nonnull final Logger log) throws IOException {
        Preconditions.checkNotNull(callback, "callback argument is required!");
        Preconditions.checkNotNull(log, "log argument is required!");
        Preconditions.checkNotNull(targetFile, "targetFile argument is required!");

        FileOutputStream fileOut = null;
        FileChannel fileChannel = null;
        try {
            final String targetFinalName = targetFile.getName();
            final File targetDirectory = targetFile.getParentFile();

            // open temporary file
            final File tmpFile = File.createTempFile(targetFinalName, ".tmp", targetDirectory);

            fileOut = new FileOutputStream(tmpFile);
            fileChannel = fileOut.getChannel();

            // make sure to use an output stream that flows THROUGH the FileChannel, so that FileChannel.force(true)
            // can do what it's supposed to

            // write the data AND flush it
            callback.writeAndFlushData(Channels.newOutputStream(fileChannel));

            return tmpFile;
        } finally {
            try {
                // fsync to disk (both data AND length)
                if (fileChannel != null) {
                    fileChannel.force(true);
                }
            } finally {
                // close the open file (if during an EXC,
                if (fileOut != null) {
                    fileOut.close();
                }
            }
        }
    }
312505
def _net_write(sock, data, expiration):
    
    current = 0
    l = len(data)
    while current < l:
        _wait_for_writable(sock, expiration)
        current += sock.send(data[current:])
394432
public void copyOutPutFilesToBucket(SubProcessConfiguration configuration, String params) {
    if (Files.exists(outFile) || Files.exists(errFile)) {
      try {
        outFileLocation = FileUtils.copyFileFromWorkerToGCS(configuration, outFile);
      } catch (Exception ex) {
        LOG.error("Error uploading log file to storage ", ex);
      }

      try {
        errFileLocation = FileUtils.copyFileFromWorkerToGCS(configuration, errFile);
      } catch (Exception ex) {
        LOG.error("Error uploading log file to storage ", ex);
      }

      LOG.info(String.format("Log Files for process: %s outFile was: %s errFile was: %s", params,
          outFileLocation, errFileLocation));
    } else {
      LOG.error(String.format("There was no output file or err file for process %s", params));
    }
  }
1212
def write(self, chunk):
        """"""
        if self.chunked_write and chunk:
            chunk_size_hex = hex(len(chunk))[2:].encode('ascii')
            buf = [chunk_size_hex, CRLF, chunk, CRLF]
            self.conn.wfile.write(EMPTY.join(buf))
        else:
            self.conn.wfile.write(chunk)
1140958
async def _write_to_container_stdin(self, write_stream, message):
        
        msg = msgpack.dumps(message, encoding="utf8", use_bin_type=True)
        self._logger.debug("Sending %i bytes to container", len(msg))
        write_stream.write(struct.pack('I', len(msg)))
        write_stream.write(msg)
        await write_stream.drain()
192157
@Override
    public String beginFileUpload() throws TException {
        String fileLoc = null;
        try {
            String path;
            String key = UUID.randomUUID().toString();
            path = StormConfig.masterInbox(conf) + "/" + key;
            FileUtils.forceMkdir(new File(path));
            FileUtils.cleanDirectory(new File(path));
            fileLoc = path + "/stormjar-" + key + ".jar";

            data.getUploaders().put(fileLoc, Channels.newChannel(new FileOutputStream(fileLoc)));
            LOG.info("Begin upload file from client to " + fileLoc);
            return path;
        } catch (FileNotFoundException e) {
            LOG.error("File not found: " + fileLoc, e);
            throw new TException(e);
        } catch (IOException e) {
            LOG.error("Upload file error: " + fileLoc, e);
            throw new TException(e);
        }
    }
201200
function createSocketInfoFile(channelName, readerUuid, paths) {
    const pathToFile = socketInfoPath(channelName, readerUuid, paths);
    return writeFile(
        pathToFile,
        JSON.stringify({
            time: microSeconds()
        })
    ).then(() => pathToFile);
}
1230648
function pipeFile (file, endCallback) {
    testPipeFile();
    testFunction(endCallback, 'endCallback');

    let readable = fs.createReadStream(file);

    readableEndCallback = endCallback;

    readable.once('error', throwAsyncError);
    readable.once('close', onReadableClose);

    readable.pipe(writable, { end: false });
}