Performance¶

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.classic.methods.HttpPut;
import org.apache.hc.client5.http.entity.EntityBuilder;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClients;
import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.hc.core5.http.HttpEntity;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import tech.onteon.falcondb.test.performance.dto.FileChunkOutputDTO;
import tech.onteon.falcondb.test.performance.dto.FileDTO;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.URI;
import java.net.URL;
import java.util.Arrays;
import java.util.Base64;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FalconDBTester {

    private final String falconDbUri;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private static PoolingHttpClientConnectionManager cm;
    private static CloseableHttpClient httpClient;

    public FalconDBTester(String falconDbUri) {
        this.falconDbUri = falconDbUri;
        cm = new PoolingHttpClientConnectionManager();
        cm.setMaxTotal(200);
        cm.setDefaultMaxPerRoute(100);
        httpClient = HttpClients.custom()
                .setConnectionManager(cm)
                .build();
    }

    /**
     * save object
     */
    public void saveObject(
            String tableName,
            String key,
            String jsonObject,
            int numberOfThreads,
            int numberOfExecutions) throws Exception {

        executeOperation(numberOfThreads, numberOfExecutions, (index) -> {
            String url = falconDbUri
                    + "/v1/saveObject"
                    + "?table-name=" + tableName
                    + "&key=" + key + "-" + index;

            HttpPost httpPost = new HttpPost(url);
            httpPost.setEntity(EntityBuilder.create().setText(jsonObject).build());

            try (CloseableHttpResponse httpResponse = httpClient.execute(httpPost)) {
                EntityUtils.consumeQuietly(httpResponse.getEntity());
            } catch (Throwable th) {
                throw new Exception("error while saving object to FalconDB", th);
            }
        });
    }

    /**
     * read object
     */
    public void readObject(
            String tableName,
            String key,
            int numberOfThreads,
            int numberOfExecutions) throws Exception {


        executeOperation(numberOfThreads, numberOfExecutions, (index) -> {
            String url = falconDbUri
                    + "/v1/readObject"
                    + "?table-name=" + tableName
                    + "&key=" + key + "-" + index
                    + "&lock=false";
            HttpGet httpGet = new HttpGet(url);

            try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) {
                HttpEntity httpEntity = httpResponse.getEntity();
                Object output = null;

                if (httpEntity != null) {
                    try (InputStream inputStream = httpEntity.getContent()) {
                        output = objectMapper.readValue(inputStream, Object.class);
                    }
                }
                EntityUtils.consumeQuietly(httpEntity);
            } catch (Throwable th) {
                throw new Exception("error while getting object from FalconDB", th);
            }
        });
    }

    /**
     * save file as bytes stream
     */
    public void saveFileAsStream(
            String fileKey,
            String fileName,
            String inputFilePath,
            int numberOfThreads,
            int numberOfExecutions) throws Exception {

        executeOperation(numberOfThreads, numberOfExecutions, (index) -> {
            HttpURLConnection httpURLConnection = null;

            try {
                httpURLConnection = (HttpURLConnection) new URL(falconDbUri + "/v2/saveBinaryFile").openConnection();
                httpURLConnection.setDoOutput(true);
                httpURLConnection.setRequestMethod("POST");
                httpURLConnection.setRequestProperty("Content-Type", "application/octet-stream");
                httpURLConnection.setRequestProperty("key", fileKey);
                httpURLConnection.setRequestProperty("fileName", fileName + "-" + index);

                try (
                        FileInputStream fileInputStream = new FileInputStream(inputFilePath);
                        OutputStream outputStream = httpURLConnection.getOutputStream()) {
                    byte[] buffer = new byte[8192];
                    int bytesRead;
                    while ((bytesRead = fileInputStream.read(buffer)) != -1) {
                        outputStream.write(buffer, 0, bytesRead);
                    }
                    outputStream.flush();

                    try (BufferedReader br = new BufferedReader(new InputStreamReader(httpURLConnection.getInputStream()))) {
                        while (br.readLine() != null) {}
                    }
                }
            } catch (Throwable th) {
                throw new Exception("error while saving file as stream to FalconDB", th);
            } finally {
                if (Objects.nonNull(httpURLConnection)) {
                    httpURLConnection.disconnect();
                }
            }
        });


    }

    /**
     * save file as base64 chunks
     */
    public void saveFileAsChunks(
            String fileKey,
            String fileName,
            int chunkSize,
            String inputFilePath,
            int numberOfThreads,
            int numberOfExecutions) throws Exception {

        executeOperation(numberOfThreads, numberOfExecutions, (index) -> {
            String url = falconDbUri + "/v1/saveOrAppendFile";

            HttpPut httpPut = new HttpPut(url);
            FileDTO fileDTO = new FileDTO();
            fileDTO.setFileName(fileName + "-" + index);
            fileDTO.setKey(fileKey);

            try (FileInputStream fis = new FileInputStream(inputFilePath)) {
                byte[] chunk = new byte[chunkSize];
                int bytesRead;
                while ((bytesRead = fis.read(chunk)) != -1) {
                    String base64Encoded = Base64.getEncoder().encodeToString(
                            bytesRead == chunkSize ? chunk : Arrays.copyOf(chunk, bytesRead)
                    );
                    fileDTO.setBase64FileContent(base64Encoded);
                    String json = objectMapper.writeValueAsString(fileDTO);

                    httpPut.setEntity(EntityBuilder.create().setText(json).build());
                    try (CloseableHttpResponse response = httpClient.execute(httpPut)) {
                        EntityUtils.consumeQuietly(response.getEntity());
                    }
                }
            } catch (Throwable th) {
                throw new Exception("error while saving file as chunks to FalconDB", th);
            }
        });
    }

    /**
     * read file as base64 chunks
     */
    public void getFileAsChunks(
            String fileKey,
            String fileName,
            int chunkSize,
            String outputFilePath,
            int numberOfThreads,
            int numberOfExecutions) throws Exception {

        executeOperation(numberOfThreads, numberOfExecutions, (index) -> {
            String url = falconDbUri + "/v1/getFileChunk"
                    + "?key=" + fileKey
                    + "&file-name=" + fileName + "-" + index
                    + "&chunk-size=" + chunkSize;

            int chunkNumber = 0;
            File outputFile = new File(outputFilePath + "-" + index);
            boolean isEOF = false;

            try (FileOutputStream fos = new FileOutputStream(outputFile)) {
                while (!isEOF) {
                    HttpGet httpGet = new HttpGet(URI.create(url + "&chunk-number=" + chunkNumber));
                    try (CloseableHttpResponse response = httpClient.execute(httpGet);
                         InputStream inputStream = response.getEntity().getContent()) {

                        FileChunkOutputDTO dto = objectMapper.readValue(inputStream, FileChunkOutputDTO.class);
                        isEOF = dto.isEOF();
                        byte[] bytes = Base64.getDecoder().decode(dto.getBase64FileContent());
                        if (dto.getReadBytesSize() != chunkSize) {
                            bytes = Arrays.copyOf(bytes, Math.max(dto.getReadBytesSize(), 0));
                        }
                        fos.write(bytes);
                    }
                    chunkNumber++;
                }
            } catch (Throwable th) {
                throw new Exception("error while getting file as chunks from FalconDB", th);
            }
        });
    }

    /**
     * read file as bytes stream
     */
    public void getFileAsStream(
            String fileKey,
            String fileName,
            int chunkSize,
            String outputFilePath,
            int numberOfThreads,
            int numberOfExecutions) throws Exception {

        executeOperation(numberOfThreads, numberOfExecutions, (index) -> {
            String url = falconDbUri + "/v1/getBinaryFile"
                    + "?key=" + fileKey
                    + "&file-name=" + fileName + "-" + index
                    + "&chunk-size=" + chunkSize;

            HttpGet httpGet = new HttpGet(url);
            File outputFile = new File(outputFilePath + "-" + index);

            try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet);
                 InputStream inputStream = httpResponse.getEntity().getContent();
                 FileOutputStream fos = new FileOutputStream(outputFile)) {

                byte[] chunk = new byte[chunkSize];
                int bytesRead;
                while ((bytesRead = inputStream.read(chunk)) != -1) {
                    fos.write(chunk, 0, bytesRead);
                }
                EntityUtils.consumeQuietly(httpResponse.getEntity());
            } catch (Throwable th) {
                throw new Exception("error while getting file as stream from FalconDB", th);
            }
        });
    }

    private void executeOperation(
            int numberOfThreads,
            int numberOfExecutions,
            IndexedRunnable runnable) throws Exception {

        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
        CountDownLatch countDownLatch = new CountDownLatch(numberOfExecutions);

        for (int i = 0; i < numberOfExecutions; i++) {
            int finalI = i;
            executorService.submit(() -> {
                try {
                    runnable.run(finalI);
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        try {
            countDownLatch.await();
        } catch (Throwable th) {
            throw new Exception("error, while executing operation on FalconDB, caused by: ", th);
        }

        executorService.shutdown();
    }

    private interface IndexedRunnable {
        void run(int index) throws Exception;
    }
}