sink part of the application (laptop)
435578
function initSink() {
  // back from server
  const socketReceive = new lfo.source.SocketReceive({
    port: 8000,
  });

  const bpfDisplay = new lfo.sink.BpfDisplay({
    width: 600,
    height: 400,
    container: $container,
    duration: 6,
    min: -180,
    max: 360,
  });

  // const loggerReceive = new lfo.sink.Logger({
  //   data: true,
  //   time: true,
  // });

  // socketReceive.connect(loggerReceive);
  socketReceive.connect(bpfDisplay);
}
1164421
def ensure_sink(self):
        """"""
        topic_info = self.pubsub.ensure_topic()
        scope, sink_path, sink_info = self.get_sink(topic_info)
        client = self.session.client('logging', 'v2', '%s.sinks' % scope)
        try:
            sink = client.execute_command('get', {'sinkName': sink_path})
        except HttpError as e:
            if e.resp.status != 404:
                raise
            sink = client.execute_command('create', sink_info)
        else:
            delta = delta_resource(sink, sink_info['body'])
            if delta:
                sink_info['updateMask'] = ','.join(delta)
                sink_info['sinkName'] = sink_path
                sink_info.pop('parent')
                sink = client.execute_command('update', sink_info)
            else:
                return sink_path

        self.pubsub.ensure_iam(publisher=sink['writerIdentity'])
        return sink_path
383007
void pathsToAttempt(Multimap<Path, String> sink, Location location) {
    Path base = fileSystem.getPath(location.getBase());

    String path = location.getPath();
    while (!path.isEmpty()) {
      String parent = path.substring(0, path.lastIndexOf('/', path.length() - 2) + 1);
      String profilePath = parent + name + ".wire";
      sink.put(base, profilePath);
      path = parent;
    }
  }
398550
def doRollover(self):
        """"""
        self.stream.close()
        try:
            if self.backupCount > 0:
                tmp_location = "%s.0" % self.baseFilename
                os.rename(self.baseFilename, tmp_location)
                for i in range(self.backupCount - 1, 0, -1):
                    sfn = "%s.%d" % (self.baseFilename, i)
                    dfn = "%s.%d" % (self.baseFilename, i + 1)
                    if os.path.exists(sfn):
                        if os.path.exists(dfn):
                            os.remove(dfn)
                        os.rename(sfn, dfn)
                dfn = self.baseFilename + ".1"
                if os.path.exists(dfn):
                    os.remove(dfn)
                os.rename(tmp_location, dfn)
        except Exception:
            pass
        finally:
            self.stream = WindowsFile(self.baseFilename, "a", self.encoding)
1209606
def application(self, v):
        """"""
        self._application = v
        self.backend.set_app(v)
1103478
function sink (msg, done) {
    var metric = msg.metric
    var as_text = JSON.stringify(metric, null, 1)

    console.log(as_text)

    done()
  }
1355007
function (application) {
    /*  determine connector filename  */
    var id = os.platform() + "-" + application;
    var cn = connectors[id];
    if (typeof cn === "undefined")
        throw new Error("unsupported platform/application combination: " + id);
    var filename = path.join(__dirname, cn);

    /*  spawn the connector as a child process  */
    this.c = spawn(filename, [], {
        stdio: [ "pipe", "pipe", process.stderr ],
        env: { "CONNECTOR": "FIXME" }
    });

    /*  set the stdin/stdout pipes to UTF-8 encoding mode  */
    this.c.stdin.setEncoding("utf8");
    this.c.stdout.setEncoding("utf8");

    /*  create line-based duplex stream out of stdin/stdout piples  */
    this.io = es.duplex(
        this.c.stdin,
        this.c.stdout.pipe(es.split(/\r?\n/))
    );

    /*  connect to the stream for capturing responses  */
    this.responses = [];
    this.io.pipe(es.through(function onData (data) {
        if (typeof data === "undefined" || data === "")
            return;
        var response = this.responses.shift();
        if (typeof response === "function")
            response(data);
    }.bind(this), function onEnd () {
        /*  currently nothing to do?!  */
    }));
}
657542
private void handleDrainedSource() throws IOException {
         if (!inCatalog() && m_committedBuffers.isEmpty()) {
             //Returning null indicates end of stream
             try {
                 if (m_pollTask != null) {
                     m_pollTask.setFuture(null);
                 }
             } catch (RejectedExecutionException reex) {
                 // Ignore, {@code GuestProcessor} was closed
             }
             m_pollTask = null;
             m_generation.onSourceDrained(m_partitionId, m_tableName);
             return;
         }
     }
1002083
def app_trim_memory(self, pid: int or str, level: str = 'RUNNING_LOW') -> None:
        '''
        '''
        _, error = self._execute('-s', self.device_sn, 'shell',
                                 'am', 'send-trim-memory', str(pid), level)
        if error and error.startswith('Error'):
            raise ApplicationsException(error.split(':', 1)[-1].strip())
431265
def remove(self, func):
        """"""
        if not self.data['name'].startswith(self.prefix):
            return
        parent = self.get_parent(self.get_log())
        _, sink_path, _ = self.get_sink()
        client = self.session.client(
            'logging', 'v2', '%s.sinks' % (parent.split('/', 1)[0]))
        try:
            client.execute_command(
                'delete', {'sinkName': sink_path})
        except HttpError as e:
            if e.resp.status != 404:
                raise
563620
public static Sink sink(File file) throws FileNotFoundException {
    if (file == null) throw new IllegalArgumentException("file == null");
    return sink(new FileOutputStream(file));
  }
540806
def clean_up(self):
        
        self.submenu.join()
        self.menu.clear_screen()
        self.menu.resume()
1301846
private void flushBuffer() throws IOException {
    if (this.dirty_) {
      if (this.diskPos_ != this.lo_) {
        super.seek(this.lo_);
      }
      int len = (int) (this.curr_ - this.lo_);
      super.write(this.buff_, 0, len);
      this.diskPos_ = this.curr_;
      this.dirty_ = false;
    }
  }
301710
def drain(self, cycles=None):
        """"""
        if not self._check_service_requirements():
            self.init_timer.stop()
            return self.finish()

        if self.anybar: self.anybar.change("orange")
        self.init_timer.stop()
        log.info("Trapping CTRL+C and starting to drain.")
        signal.signal(signal.SIGINT, self._handle_ctrl_c)
        with ignored(KeyboardInterrupt):
            return self._drain(cycles)
763388
function(req, res, next) {
        async.parallel({
            applications: (fn) => {
                const spy = { stash: {} };
                applications.get(req, spy, () => {
                    if(spy.stash.code !== 200) {
                        return fn(new Error('Error getting applications.'));
                    }

                    return fn(null, spy.stash.body);
                });
            },
            hosts: (fn) => {
                const spy = { stash: {} };
                hosts.get(req, spy, () => {
                    if(spy.stash.code !== 200) {
                        return fn(new Error('Error getting hosts.'));
                    }

                    return fn(null, spy.stash.body);
                });
            }
        }, (err, results) => {
            if(err) {
                res.stash.code = 500;
                return next();
            }

            res.stash.body = {
                applications: results.applications,
                hosts: results.hosts
            };
            res.stash.code = 200;
            return next();
        });
    }
1143699
@GuardedBy("evictionLock")
    void drainReadBuffers() {
        final int start = (int) Thread.currentThread().getId();
        final int end = start + NUMBER_OF_READ_BUFFERS;
        for (int i = start; i < end; i++) {
            drainReadBuffer(i & READ_BUFFERS_MASK);
        }
    }
650017
def cleanup(test_data, udfs, tmp_data, tmp_db):
    """"""
    con = make_ibis_client(ENV)

    if udfs:
        # this comes before test_data bc the latter clobbers this too
        con.hdfs.rmdir(os.path.join(ENV.test_data_dir, 'udf'))

    if test_data:
        con.drop_database(ENV.test_data_db, force=True)
        con.hdfs.rmdir(ENV.test_data_dir)

    if tmp_data:
        con.hdfs.rmdir(ENV.tmp_dir)

    if tmp_db:
        con.drop_database(ENV.tmp_db, force=True)