标签: stream

我正在从 BlockBlob 读取文件。文件大小保证小于 64 MB。因此它是单块操作。读取文件后，我更改了其中的某些部分，并通过 CloudBlockBlob 的 UploadFromStream 函数重新上传。我的问题是“对于小于 64 MB 的大小，CloudBlockBlob 的 UploadFromStream 函数是否是原子的？”。在写入过程中发生异常后，是否有可能在天蓝色存储上有损坏的文件？

注意：我对 AppendBlobs 提出了类似的问题，并得到了它是原子的答案。

我现在有一个统计和哈希（可能很大）文件的流程。现在我正在这样做：

function gen_hash (fn, callback) {
  var hash = crypto.createHash('sha256');

  var fh = fs.createReadStream(fn);
  fh.on('data', d => hash.update(d));
  fh.on('end', () => {
    var digest = hash.digest('hex');
  });
}

function gen_stat (fn, callback) {
  fs.stat(fn, function (err, stats) {
    if (err) {
      if (err.code == "ENOENT") {
        file.exists = false;
      }
    }
  }
}

然后我有一些代码允许在它们都完成时进行回调。正如你可以想象的那样，它相当……复杂。

我认为这与 Promise 非常匹配，但我不知道如何使用它们。

我尝试了一些类似的东西：

const Promises = require('bluebird');
const fs = Promises.promisifyAll(require('fs'));

fh = fs.createReadStream(fn)
.then...
.error...

但我实际上不明白这里要做什么，而且蓝鸟网站似乎缺乏细节。

我觉得类似的东西Promise.all可能是正确的，但我只是没有看到任何正确语法的好例子。

我知道我可以编写一个包装函数，在各种条件下返回一个承诺，但我不太明白它在承诺领域是如何工作的。也许是这样的？

const fs = Promise.promisifyAll(require('fs')); …

我想对流执行一些操作，然后将流分成两个流，然后分别处理它们。

显示问题的示例：

Flowable<SuccessfulObject> stream = Flowable.fromArray(
        new SuccessfulObject(true, 0),
        new SuccessfulObject(false, 1),
        new SuccessfulObject(true, 2));

stream = stream.doOnEach(System.out::println);

Flowable<SuccessfulObject> successful = stream.filter(SuccessfulObject::isSuccess);
Flowable<SuccessfulObject> failed = stream.filter(SuccessfulObject::isFail);

successful.doOnEach(successfulObject -> {/*handle success*/}).subscribe();
failed.doOnEach(successfulObject -> {/*handle fail*/}).subscribe();

班级：

class SuccessfulObject {
    private boolean success;
    private int id;

    public SuccessfulObject(boolean success, int id) {
        this.success = success;
        this.id = id;
    }

    public boolean isSuccess() {
        return success;
    }
    public boolean isFail() {
        return !success;
    }

    public void setSuccess(boolean success) {
        this.success = success;
    } …

我试图通过删除消费者配置并让我的脚本稍后重新创建它来重置消费者配置，但我遇到了关于新消费者不可删除的错误。

kafka@kafka-0:~$ ./bin/kafka-consumer-groups.sh --bootstrap kafka-0:9092 --delete --group etl
Option '[delete]' is only valid with '[zookeeper]'. Note that there's no need to delete group metadata for the new consumer as the group is deleted when the last committed offset for that group expires.

现在我想知道，控制此错误消息过期的消费者配置选项的名称是什么？

我正在尝试将一个对象写入流：

const fs = require('fs');

const stream = fs.createWriteStream('index.json', {objectMode: true});
stream.write({"a":1});

stream.end();

根据文档，我可以通过传递选项来做到这一点{objectMode: true}，但我收到错误：

events.js:182
      throw er; // Unhandled 'error' event
      ^

Error: Invalid data
    at WriteStream._write (fs.js:2116:31)
    at doWrite (_stream_writable.js:385:12)
    at writeOrBuffer (_stream_writable.js:371:5)
    at WriteStream.Writable.write (_stream_writable.js:288:11)
    at Object.<anonymous> (/Users/rostik404/projects/test-project/test-stream.js:8:8)
    at Module._compile (module.js:573:30)
    at Object.Module._extensions..js (module.js:584:10)
    at Module.load (module.js:507:32)
    at tryModuleLoad (module.js:470:12)
    at Function.Module._load (module.js:462:3)

节点版本是8.4.0

您好，我已经编写了语句，如果参数存在并且其值为falseOptional<Parameter>，我将返回true。

public boolean getNoDailyAllowance(String code) {
    Optional<Parameter> myParam = parameterDao.getCachedParameter(code);
    return myParam.isPresent() && !myParam.get().currentValueBoolean();
}

return calcDailyAllowanceParam.map(parameter -> Boolean.parseBoolean(parameter.getCurrentValue())).orElse(false);