private static Flux<DataBuffer> readFile(Vertx vertx, Path path) {
    AsyncFile file = vertx.fileSystem().openBlocking(path.toString(), new OpenOptions());

    Flux<Buffer> buffers = Flux.create(sink -> {
        file.pause();
        file.endHandler(v -> sink.complete());
        file.exceptionHandler(sink::error);
        file.handler(sink::next);
        sink.onRequest(file::fetch);
    });

    DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();

    return Flux.from(buffers)
        .map(Buffer::getBytes)
        .map(dataBufferFactory::wrap);
}
 

/**
 * Fake endpoint for testing various parameters 假端點 偽のエンドポイント 가짜 엔드 포인트 
 * Fake endpoint for testing various parameters 假端點 偽のエンドポイント 가짜 엔드 포인트 
 *
 * @param context Vertx test context for doing assertions
 */
@Test
public void testEndpointParametersTest(TestContext context) {
    Async async = context.async();
    BigDecimal number = null;
    Double _double = null;
    String patternWithoutDelimiter = null;
    byte[] _byte = null;
    Integer integer = null;
    Integer int32 = null;
    Long int64 = null;
    Float _float = null;
    String string = null;
    AsyncFile binary = null;
    LocalDate date = null;
    OffsetDateTime dateTime = null;
    String password = null;
    String paramCallback = null;
    api.testEndpointParameters(number, _double, patternWithoutDelimiter, _byte, integer, int32, int64, _float, string, binary, date, dateTime, password, paramCallback, result -> {
        // TODO: test validations
        async.complete();
    });
}
 

private void downloadFile(AsyncFile file, HttpServerRequest request) {
  HttpServerResponse response = request.response();
  response.setStatusCode(200)
    .putHeader("Content-Type", "audio/mpeg")
    .setChunked(true);

  file.handler(buffer -> {
    response.write(buffer);
    if (response.writeQueueFull()) {
      file.pause();
      response.drainHandler(v -> file.resume());
    }
  });

  file.endHandler(v -> response.end());
}
 

public static void main(String[] args) {
  Vertx vertx = Vertx.vertx();
  OpenOptions opts = new OpenOptions().setRead(true);
  vertx.fileSystem().open("build.gradle.kts", opts, ar -> {
    if (ar.succeeded()) {
      AsyncFile file = ar.result();
      file.handler(System.out::println)
        .exceptionHandler(Throwable::printStackTrace)
        .endHandler(done -> {
          System.out.println("\n--- DONE");
          vertx.close();
        });
    } else {
      ar.cause().printStackTrace();
    }
  });
}
 

protected void onFileOpened(File file, AsyncResult<AsyncFile> ar, CompletableFuture<File> future) {
  if (ar.failed()) {
    future.completeExceptionally(ar.cause());
    return;
  }

  AsyncFile asyncFile = ar.result();
  CompletableFuture<Void> saveFuture = saveToWriteStream(asyncFile);
  saveFuture.whenComplete((v, saveException) -> {
    asyncFile.close(closeAr -> {
      if (closeAr.failed()) {
        LOGGER.error("Failed to close file {}.", file);
      }

      // whatever close success or failed
      // will not affect to result
      // result just only related to write
      if (saveException == null) {
        future.complete(file);
        return;
      }

      future.completeExceptionally(saveException);
    });
  });
}
 

@Test
public void copyZeroLength(TestContext context) throws IOException {
    runOnServerContext(context, () -> {
        Path tmpFile = Files.createTempFile(tmpDir(), "", "");

        AsyncFile asyncFile = vertx().fileSystem().openBlocking(tmpFile.toString(), new OpenOptions());
        final BufferWriteEndableWriteStream bufferWriteStream = new BufferWriteEndableWriteStream();

        return AsyncIO.pump(EndableReadStream.from(asyncFile), bufferWriteStream)
                .map(new Func1<Void, Void>() {
                    @Override
                    public Void call(Void aVoid) {
                        VertxAssert.assertEquals(context, 0, bufferWriteStream.toBuffer().length());
                        return null;
                    }
                });
    });
}
 

@Test
public void testMutateCodecInterceptor() throws Exception {
  server.requestHandler(req -> req.response().end("foo!"));
  startServer();
  File f = Files.createTempFile("vertx", ".dat").toFile();
  assertTrue(f.delete());
  AsyncFile foo = vertx.fileSystem().openBlocking(f.getAbsolutePath(), new OpenOptions().setSync(true).setTruncateExisting(true));
  client.addInterceptor(this::handleMutateCodec);
  HttpRequest<Void> builder = client.get("/somepath").as(BodyCodec.pipe(foo));
  builder.send(onSuccess(resp -> {
    foo.write(Buffer.buffer("bar!"));
    foo.close(onSuccess(v -> {
      assertEquals("bar!", vertx.fileSystem().readFileBlocking(f.getAbsolutePath()).toString());
      testComplete();
    }));
  }));
  await();
  if (f.exists()) {
    f.delete();
  }
}
 

/**
     * uploads an image
     * 
     * @param petId ID of pet to update (required)
     * @param additionalMetadata Additional data to pass to server (optional)
     * @param file file to upload (optional)
     * @param resultHandler Asynchronous result handler
     */
    public void uploadFile(Long petId, String additionalMetadata, AsyncFile file, Handler<AsyncResult<ModelApiResponse>> resultHandler) {
        Object localVarBody = null;
        
        // verify the required parameter 'petId' is set
        if (petId == null) {
            resultHandler.handle(ApiException.fail(400, "Missing the required parameter 'petId' when calling uploadFile"));
            return;
        }
        
        // create path and map variables
        String localVarPath = "/pet/{petId}/uploadImage".replaceAll("\\{" + "petId" + "\\}", petId.toString());

        // query params
        List<Pair> localVarQueryParams = new ArrayList<>();

        // header params
        MultiMap localVarHeaderParams = MultiMap.caseInsensitiveMultiMap();
        
        // cookie params
        MultiMap localVarCookieParams = MultiMap.caseInsensitiveMultiMap();
        
        // form params
        // TODO: sending files within multipart/form-data is not supported yet (because of vertx web-client)
        Map<String, Object> localVarFormParams = new HashMap<>();
        if (additionalMetadata != null) localVarFormParams.put("additionalMetadata", additionalMetadata);
if (file != null) localVarFormParams.put("file", file);

        String[] localVarAccepts = { "application/json" };
        String[] localVarContentTypes = { "multipart/form-data" };
        String[] localVarAuthNames = new String[] { "petstore_auth" };
        TypeReference<ModelApiResponse> localVarReturnType = new TypeReference<ModelApiResponse>() {};
        apiClient.invokeAPI(localVarPath, "POST", localVarQueryParams, localVarBody, localVarHeaderParams, localVarCookieParams, localVarFormParams, localVarAccepts, localVarContentTypes, localVarAuthNames, localVarReturnType, resultHandler);
    }
 

/**
     * uploads an image (required)
     * 
     * @param petId ID of pet to update (required)
     * @param requiredFile file to upload (required)
     * @param additionalMetadata Additional data to pass to server (optional)
     * @param resultHandler Asynchronous result handler
     */
    public void uploadFileWithRequiredFile(Long petId, AsyncFile requiredFile, String additionalMetadata, Handler<AsyncResult<ModelApiResponse>> resultHandler) {
        Object localVarBody = null;
        
        // verify the required parameter 'petId' is set
        if (petId == null) {
            resultHandler.handle(ApiException.fail(400, "Missing the required parameter 'petId' when calling uploadFileWithRequiredFile"));
            return;
        }
        
        // verify the required parameter 'requiredFile' is set
        if (requiredFile == null) {
            resultHandler.handle(ApiException.fail(400, "Missing the required parameter 'requiredFile' when calling uploadFileWithRequiredFile"));
            return;
        }
        
        // create path and map variables
        String localVarPath = "/fake/{petId}/uploadImageWithRequiredFile".replaceAll("\\{" + "petId" + "\\}", petId.toString());

        // query params
        List<Pair> localVarQueryParams = new ArrayList<>();

        // header params
        MultiMap localVarHeaderParams = MultiMap.caseInsensitiveMultiMap();
        
        // cookie params
        MultiMap localVarCookieParams = MultiMap.caseInsensitiveMultiMap();
        
        // form params
        // TODO: sending files within multipart/form-data is not supported yet (because of vertx web-client)
        Map<String, Object> localVarFormParams = new HashMap<>();
        if (additionalMetadata != null) localVarFormParams.put("additionalMetadata", additionalMetadata);
if (requiredFile != null) localVarFormParams.put("requiredFile", requiredFile);

        String[] localVarAccepts = { "application/json" };
        String[] localVarContentTypes = { "multipart/form-data" };
        String[] localVarAuthNames = new String[] { "petstore_auth" };
        TypeReference<ModelApiResponse> localVarReturnType = new TypeReference<ModelApiResponse>() {};
        apiClient.invokeAPI(localVarPath, "POST", localVarQueryParams, localVarBody, localVarHeaderParams, localVarCookieParams, localVarFormParams, localVarAccepts, localVarContentTypes, localVarAuthNames, localVarReturnType, resultHandler);
    }
 

public void sendBody(HttpRequest<Buffer> request,
                     Handler<AsyncResult<HttpResponse<Buffer>>> responseHandler,
                     Object body) {
    if (body instanceof byte[]) {
        Buffer buffer = Buffer.buffer((byte[]) body);
        request.sendBuffer(buffer, responseHandler);
    } else if (body instanceof AsyncFile) {
        AsyncFile file = (AsyncFile) body;
        request.sendStream(file, responseHandler);
    } else {
        request.sendJson(body, responseHandler);
    }
}
 

/**
 * File Download handling.
 *
 * @param response The HTTP response
 * @param handler The response handler
 */
protected <T> void handleFileDownload(HttpResponse<Buffer> response, Handler<AsyncResult<T>> handler) {
    FileSystem fs = getVertx().fileSystem();

    String filename = generateFilename(response.headers());
    Consumer<String> fileHandler = directory -> {
        fs.open(directory + filename, FILE_DOWNLOAD_OPTIONS, asyncFileResult -> {
            if (asyncFileResult.succeeded()) {
                AsyncFile asyncFile = asyncFileResult.result();
                asyncFile.write(response.bodyAsBuffer());
                //noinspection unchecked
                handler.handle(Future.succeededFuture((T) asyncFile));
            } else {
                handler.handle(ApiException.fail(asyncFileResult.cause()));
            }
        });
    };

    String dir = getDownloadsDir();
    if (dir != null && !dir.isEmpty()) {
        fs.mkdirs(dir, mkdirResult -> {
            String sanitizedFolder = dir.endsWith("/") ? dir : dir + "/";
            fileHandler.accept(sanitizedFolder);
        });
    } else {
        fileHandler.accept("");
    }
}
 

/**
 * Build a response handler for the HttpResponse.
 *
 * @param returnType The return type
 * @param handler The response handler
 * @return The HTTP response handler
 */
protected <T> Handler<AsyncResult<HttpResponse<Buffer>>> buildResponseHandler(TypeReference<T> returnType,
                                                                              Handler<AsyncResult<T>> handler) {
    return response -> {
        AsyncResult<T> result;
        if (response.succeeded()) {
            HttpResponse<Buffer> httpResponse = response.result();
            if (httpResponse.statusCode() / 100 == 2) {
                if (httpResponse.statusCode() == 204 || returnType == null) {
                    result = Future.succeededFuture(null);
                } else {
                    T resultContent = null;
                    if ("byte[]".equals(returnType.getType().toString())) {
                        resultContent = (T) httpResponse.body().getBytes();
                        result = Future.succeededFuture(resultContent);
                    } else if (AsyncFile.class.equals(returnType.getType())) {
                        handleFileDownload(httpResponse, handler);
                        return;
                    } else {
                        try {
                            resultContent = this.objectMapper.readValue(httpResponse.bodyAsString(), returnType);
                            result = Future.succeededFuture(resultContent);
                        } catch (Exception e) {
                            result =  ApiException.fail(new DecodeException("Failed to decode:" + e.getMessage(), e));
                        }
                    }
                }
            } else {
                result = ApiException.fail(httpResponse.statusMessage(), httpResponse.statusCode(), httpResponse.headers(), httpResponse.bodyAsString());
            }
        } else if (response.cause() instanceof ApiException) {
            result = Future.failedFuture(response.cause());
        } else {
            result = ApiException.fail(500, response.cause() != null ? response.cause().getMessage() : null);
        }
        handler.handle(result);
    };
}
 

/**
 * Get binary
 * @return binary
**/
@javax.annotation.Nullable
@ApiModelProperty(value = "")
@JsonProperty(JSON_PROPERTY_BINARY)
@JsonInclude(value = JsonInclude.Include.USE_DEFAULTS)

public AsyncFile getBinary() {
  return binary;
}
 

/**
 * uploads an image
 * 
 *
 * @param context Vertx test context for doing assertions
 */
@Test
public void uploadFileTest(TestContext context) {
    Async async = context.async();
    Long petId = null;
    String additionalMetadata = null;
    AsyncFile file = null;
    api.uploadFile(petId, additionalMetadata, file, result -> {
        // TODO: test validations
        async.complete();
    });
}
 

public static void main(String[] args) {
  Vertx vertx = Vertx.vertx();

  AsyncFile file = vertx.fileSystem().openBlocking("sample.db",
    new OpenOptions().setRead(true));

  RecordParser parser = RecordParser.newFixed(4, file);
  parser.handler(header -> readMagicNumber(header, parser));
  parser.endHandler(v -> vertx.close());
}
 

public static void main(String[] args) {
  Vertx vertx = Vertx.vertx();

  AsyncFile file = vertx.fileSystem().openBlocking("sample.db",
    new OpenOptions().setRead(true));

  RecordParser parser = RecordParser.newFixed(4, file);
  parser.pause();
  parser.fetch(1);
  parser.handler(header -> readMagicNumber(header, parser));
  parser.endHandler(v -> vertx.close());
}
 

public static void main(String[] args) {
  Vertx vertx = Vertx.vertx();
  AsyncFile file = vertx.fileSystem().openBlocking("sample.db",
    new OpenOptions().setWrite(true).setCreate(true));

  Buffer buffer = Buffer.buffer();

  // Magic number
  buffer.appendBytes(new byte[] { 1, 2, 3, 4});

  // Version
  buffer.appendInt(2);

  // DB name
  buffer.appendString("Sample database\n");

  // Entry 1
  String key = "abc";
  String value = "123456-abcdef";
  buffer
    .appendInt(key.length())
    .appendString(key)
    .appendInt(value.length())
    .appendString(value);

  // Entry 2
  key = "[email protected]";
  value = "Foo Bar Baz";
  buffer
    .appendInt(key.length())
    .appendString(key)
    .appendInt(value.length())
    .appendString(value);

  file.end(buffer, ar -> vertx.close());
}
 

private void downloadFilePipe(AsyncFile file, HttpServerRequest request) {
  HttpServerResponse response = request.response();
  response.setStatusCode(200)
    .putHeader("Content-Type", "audio/mpeg")
    .setChunked(true);
  file.pipeTo(response);
}
 

private void handleFileOpenWithDownload(AsyncResult<AsyncFile> openResult, Promise<Void> promise) {
    if (openResult.succeeded()) {
        final AsyncFile asyncFile = openResult.result();
        try {
            httpClient.getAbs(downloadUrl, response -> pumpFileFromRequest(response, asyncFile, promise)).end();
        } catch (Exception e) {
            promise.fail(e);
        }
    } else {
        promise.fail(openResult.cause());
    }
}
 

private void pumpFileFromRequest(
        HttpClientResponse httpClientResponse, AsyncFile asyncFile, Promise<Void> promise) {

    logger.info("Trying to download file from {0}", downloadUrl);
    httpClientResponse.pause();
    final Pump pump = Pump.pump(httpClientResponse, asyncFile);
    pump.start();
    httpClientResponse.resume();

    final long idTimer = setTimeoutTimer(asyncFile, pump, promise);

    httpClientResponse.endHandler(responseEndResult -> handleResponseEnd(asyncFile, idTimer, promise));
}
 

private void handleTimeout(AsyncFile asyncFile, Pump pump, Promise<Void> promise) {
    pump.stop();
    asyncFile.close();
    if (!promise.future().isComplete()) {
        promise.fail(new TimeoutException("Timeout on download"));
    }
}
 

public static Observable<Void> close(AsyncFile asyncFile) {
    try {
        ObservableFuture<Void> rh = RxHelper.observableFuture();
        asyncFile.close(rh.toHandler());
        return rh;
    } catch (Throwable e) {
        return Observable.error(e);
    }
}
 

@Test
public void testMetadataAndReadStream(TestContext context) throws IOException {

    byte[] data = new byte[256 * 1024 * 1024];
    getCurrentInstance().nextBytesBlocking(data);

    Path dataFile = path.resolve(".data");
    write(dataFile, data, CREATE_NEW);

    long size = size(dataFile);
    final byte[] expectedDataSha512 = hash(dataFile.toFile(), sha512()).asBytes();
    final AsyncFile bigFile = sfsVertx.fileSystem().openBlocking(dataFile.toString(), new OpenOptions());

    Path journalPath = path.resolve(".journal");
    JournalFile journalFile = new JournalFile(journalPath);

    Async async = context.async();

    aVoid()
            .flatMap(aVoid -> journalFile.open(sfsVertx))
            .flatMap(aVoid -> journalFile.enableWrites(sfsVertx))
            .flatMap(aVoid -> journalFile.append(sfsVertx, buffer("metadata0", UTF_8.toString()), size, EndableReadStream.from(bigFile)))
            .doOnNext(aVoid -> bigFile.setReadPos(0))
            .flatMap(aVoid -> journalFile.append(sfsVertx, buffer("metadata1", UTF_8.toString()), size, EndableReadStream.from(bigFile)))
            .doOnNext(aVoid -> bigFile.setReadPos(0))
            .flatMap(aVoid -> journalFile.append(sfsVertx, buffer("metadata2", UTF_8.toString()), size, EndableReadStream.from(bigFile)))
            .doOnNext(aVoid -> bigFile.setReadPos(0))
            .flatMap(aVoid -> journalFile.append(sfsVertx, buffer("metadata3", UTF_8.toString()), size, EndableReadStream.from(bigFile)))
            .doOnNext(aVoid -> bigFile.setReadPos(0))
            .flatMap(aVoid -> journalFile.append(sfsVertx, buffer("metadata4", UTF_8.toString()), size, EndableReadStream.from(bigFile)))
            // assert stuff before closing
            .flatMap(aVoid -> assertScanDataReadStream(context, sfsVertx, journalFile, 5, "metadata", expectedDataSha512))
            .flatMap(aVoid -> journalFile.disableWrites(sfsVertx))
            .flatMap(aVoid -> journalFile.force(sfsVertx, true))
            .flatMap(aVoid -> journalFile.close(sfsVertx))
            // assert stuff can be read closing and opening
            .flatMap(aVoid -> journalFile.open(sfsVertx))
            .flatMap(aVoid -> assertScanDataReadStream(context, sfsVertx, journalFile, 5, "metadata", expectedDataSha512))
            .subscribe(new TestSubscriber(context, async));
}
 

@Test
public void testImmediatePumpFile(TestContext context) throws IOException {
    SfsVertx vertx = new SfsVertxImpl(rule.vertx(), backgroundPool, ioPool);

    byte[] bytes = new byte[256];
    getCurrentInstance().nextBytesBlocking(bytes);
    Path path = createTempFile("", "");

    try (OutputStream outputStream = newOutputStream(path)) {
        for (int i = 0; i < 10000; i++) {
            outputStream.write(bytes);
        }
    }

    final byte[] sha512 = hash(path.toFile(), sha512()).asBytes();

    Async async = context.async();
    aVoid()
            .flatMap(aVoid -> {
                AsyncFile asyncFile = vertx.fileSystem().openBlocking(path.toString(), new OpenOptions());
                PipedReadStream pipedReadStream = new PipedReadStream();
                PipedEndableWriteStream pipedEndableWriteStream = new PipedEndableWriteStream(pipedReadStream);
                Observable<Void> producer = pump(EndableReadStream.from(asyncFile), pipedEndableWriteStream);
                DigestEndableWriteStream digestWriteStream = new DigestEndableWriteStream(new NullEndableWriteStream(), SHA512);
                Observable<Void> consumer = pump(pipedReadStream, digestWriteStream);
                return combineSinglesDelayError(producer, consumer, (aVoid1, aVoid2) -> {
                    assertArrayEquals(context, sha512, digestWriteStream.getDigest(SHA512).get());
                    return (Void) null;
                });
            })
            .doOnTerminate(() -> {
                try {
                    deleteIfExists(path);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            })
            .subscribe(new TestSubscriber(context, async));
}
 

public void toObservable(Vertx vertx) {
  FileSystem fileSystem = vertx.fileSystem();
  fileSystem.open("/data.txt", new OpenOptions(), result -> {
    AsyncFile file = result.result();
    Observable<Buffer> observable = RxHelper.toObservable(file);
    observable.forEach(data -> System.out.println("Read data: " + data.toString("UTF-8")));
  });
}
 

public void unmarshaller(FileSystem fileSystem) {
  fileSystem.open("/data.txt", new OpenOptions(), result -> {
    AsyncFile file = result.result();
    Observable<Buffer> observable = RxHelper.toObservable(file);
    observable.lift(RxHelper.unmarshaller(MyPojo.class)).subscribe(
        mypojo -> {
          // Process the object
        }
    );
  });
}
 

public void toFlowable(Vertx vertx) {
  FileSystem fileSystem = vertx.fileSystem();
  fileSystem.open("/data.txt", new OpenOptions(), result -> {
    AsyncFile file = result.result();
    Flowable<Buffer> observable = FlowableHelper.toFlowable(file);
    observable.forEach(data -> System.out.println("Read data: " + data.toString("UTF-8")));
  });
}
 

public void unmarshaller(FileSystem fileSystem) {
  fileSystem.open("/data.txt", new OpenOptions(), result -> {
    AsyncFile file = result.result();
    Flowable<Buffer> observable = FlowableHelper.toFlowable(file);
    observable.compose(FlowableHelper.unmarshaller(MyPojo.class)).subscribe(
        mypojo -> {
          // Process the object
        }
    );
  });
}
 

public void example30(MongoGridFsClient gridFsStreamClient, AsyncFile asyncFile) {
  GridFsUploadOptions options = new GridFsUploadOptions();
  options.setChunkSizeBytes(2048);
  options.setMetadata(new JsonObject().put("catagory", "Polynesian gods"));
  gridFsStreamClient.uploadByFileNameWithOptions(asyncFile, "kanaloa", options, stringAsyncResult -> {
    String id = stringAsyncResult.result();
  });

}
 

public void example32(MongoGridFsClient gridFsStreamClient, AsyncFile asyncFile) {
  GridFsDownloadOptions options = new GridFsDownloadOptions();
  options.setRevision(0);
  gridFsStreamClient.downloadByFileNameWithOptions(asyncFile, "kamapuaa.fil", options, longAsyncResult -> {
    Long length = longAsyncResult.result();
  });
}