smartbucket/ts/smartbucket.classes.bucket.ts

import * as plugins from './smartbucket.plugins';
import { SmartBucket } from './smartbucket.classes.smartbucket';
import { Directory } from './smartbucket.classes.directory';
import { Observable } from 'rxjs';

export class Bucket {
  public static async getBucketByName(smartbucketRef: SmartBucket, bucketNameArg: string) {
    const buckets = await smartbucketRef.minioClient.listBuckets();
    const foundBucket = buckets.find((bucket) => {
      return bucket.name === bucketNameArg;
    });

    if (foundBucket) {
      console.log(`bucket with name ${bucketNameArg} exists.`);
      console.log(`Taking this as base for new Bucket instance`);
      return new this(smartbucketRef, bucketNameArg);
    } else {
      return null;
    }
  }

  public static async createBucketByName(smartbucketRef: SmartBucket, bucketName: string) {
    await smartbucketRef.minioClient.makeBucket(bucketName, 'ams3').catch((e) => console.log(e));
    return new Bucket(smartbucketRef, bucketName);
  }

  public static async removeBucketByName(smartbucketRef: SmartBucket, bucketName: string) {
    await smartbucketRef.minioClient.removeBucket(bucketName).catch((e) => console.log(e));
  }

  public smartbucketRef: SmartBucket;
  public name: string;

  constructor(smartbucketRef: SmartBucket, bucketName: string) {
    this.smartbucketRef = smartbucketRef;
    this.name = bucketName;
  }

  /**
   * gets the base directory of the bucket
   */
  public async getBaseDirectory() {
    return new Directory(this, null, '');
  }

  // ===============
  // Fast Operations
  // ===============

  /**
   * store file
   */
  public async fastStore(pathArg: string, fileContent: string | Buffer): Promise<void> {
    const streamIntake = new plugins.streamfunction.Intake();
    const putPromise = this.smartbucketRef.minioClient
      .putObject(this.name, pathArg, streamIntake.getReadable())
      .catch((e) => console.log(e));
    streamIntake.pushData(fileContent);
    streamIntake.signalEnd();
    await putPromise;
  }

  /**
   * get file
   */
  public async fastGet(pathArg: string): Promise<Buffer> {
    const done = plugins.smartpromise.defer();
    let completeFile: Buffer;
    const replaySubject = await this.fastGetStream(pathArg);
    const subscription = replaySubject.subscribe(
      (chunk) => {
        if (completeFile) {
          completeFile = Buffer.concat([completeFile, chunk]);
        }
      },
      (err) => {
        console.log(err);
      },
      () => {
        done.resolve();
        subscription.unsubscribe();
      }
    );
    await done.promise;
    return completeFile;
  }

  public async fastGetStream(pathArg: string): Promise<plugins.smartrx.rxjs.ReplaySubject<Buffer>> {
    const fileStream = await this.smartbucketRef.minioClient
      .getObject(this.name, pathArg)
      .catch((e) => console.log(e));
    const replaySubject = new plugins.smartrx.rxjs.ReplaySubject<Buffer>();
    const duplexStream = plugins.streamfunction.createDuplexStream<Buffer, Buffer>(
      async (chunk) => {
        replaySubject.next(chunk);
        return chunk;
      },
      async (cb) => {
        replaySubject.complete();
        return Buffer.from('');
      }
    );

    if (!fileStream) {
      return null;
    }

    fileStream.pipe(duplexStream);
    return replaySubject;
  }

  /**
   * removeObject
   */
  public async fastRemove(pathArg: string) {
    await this.smartbucketRef.minioClient.removeObject(this.name, pathArg);
  }
}