def consume_batch_async(self, batch):
        op = self.operation()
        op_mutate = op in ['set', 'add']

        stdout = sys.stdout
        msg_visitor = None

        opts_etc = getattr(self.opts, "etc", None)
        if opts_etc:
            stdout = opts_etc.get("stdout", sys.stdout)
            msg_visitor = opts_etc.get("msg_visitor", None)

        for msg in batch.msgs:
            if msg_visitor:
                msg = msg_visitor(msg)

            cmd, vbucket_id, key, flg, exp, cas, meta, val = msg
            if self.skip(key, vbucket_id):
                continue

            try:
                if cmd == couchbaseConstants.CMD_TAP_MUTATION:
                    if op_mutate:
                        # <op> <key> <flags> <exptime> <bytes> [noreply]\r\n
                        stdout.write("%s %s %s %s %s\r\n" %
                                     (op, key, flg, exp, len(val)))
                        stdout.write(val)
                        stdout.write("\r\n")
                    elif op == 'get':
                        stdout.write("get %s\r\n" % (key))
                elif cmd == couchbaseConstants.CMD_TAP_DELETE:
                    if op_mutate:
                        stdout.write("delete %s\r\n" % (key))
                elif cmd == couchbaseConstants.CMD_GET:
                    stdout.write("get %s\r\n" % (key))
                else:
                    return "error: StdOutSink - unknown cmd: " + str(cmd), None
            except __HOLE__:
                return "error: could not write to stdout", None

        future = SinkBatchFuture(self, batch)
        self.future_done(future, 0)
        return 0, future


# --------------------------------------------------
dataset/ETHPy150Open membase/membase-cli/pump.py/StdOutSink.consume_batch_async