datastore.js

'use strict';

var Lock = require('lock');
var stream = require('stream');
var bl = require('bl');
var Block = require('ipfs-block');
var fs = require('fs')
var join = require('path').join
var mkdirp = require('mkdirp')

var PREFIX_LENGTH = 8;

exports = module.exports;

function multihashToPath(multihash, extension) {
  extension = extension || 'data';
  var filename = multihash.toString('hex') + '.' + extension;
  var folder = filename.slice(0, PREFIX_LENGTH);
  var path = folder + '/' + filename;

  return path;
}

exports.setUp = function (basePath, blobStore, locks) {
  var store = blobStore(basePath + '/blocks');
  var lock = new Lock();

  var createReadStream = function createReadStream(multihash, extension) {
    var path = multihashToPath(multihash, extension);
    return store.createReadStream(path);
  };

  var readFile = function (mh, ext, cb) {
    var path = join(basePath, 'blocks', multihashToPath(mh, ext));

    lock(path, function (release) {
      fs.readFile(path, release(cb))
    })

  }

  var createWriteStream = function createWriteStream(multihash, extension, cb) {
    var path = multihashToPath(multihash, extension);
    var through = stream.PassThrough();

    lock(path, function (release) {
      var ws = store.createWriteStream(path, release(cb));
      through.pipe(ws);
    });

    return through;
  };

  var writeFile = function (mh, ext, data, cb) {
    var path = join(basePath, 'blocks', multihashToPath(mh, ext));
    var folder = path.split('/').slice(0, -1).join('/')
    mkdirp(folder, function (err) {
      if (err) return cb(err)
      lock(path, function (release) {
        console.log('writing to %s', path)
        fs.writeFile(path, data, release(cb))
      });
    })
  }

  return {
    get: function get(key, extension, cb) {
      if (typeof extension === 'function') {
        cb = extension;
        extension = 'data';
      }

      if (!key) {
        return cb(new Error('Invalid key'));
      }
      readFile(key, extension, (err, data) => {
      // createReadStream(key, extension).pipe(bl(function (err, data) {
        if (err) {
          return cb(err);
        }
        if (extension === 'data') {
          extension = 'protobuf';
        }
        console.log('received', data)
        cb(null, new Block(data, extension));
      })//);
    },

    put: function put(block, cb) {
      if (!block || !block.data) {
        return cb(new Error('Invalid block'));
      }

      // var ws = createWriteStream(block.key, block.extension, cb);
      // ws.write(block.data);
      // ws.end();
      writeFile(block.key, block.extension, block.data, cb)
    },

    has: function has(key, extension, cb) {
      if (typeof extension === 'function') {
        cb = extension;
        extension = undefined;
      }

      if (!key) {
        return cb(new Error('Invalid key'));
      }

      var path = multihashToPath(key, extension);
      store.exists(path, cb);
    },

    delete: function _delete(key, extension, cb) {
      if (typeof extension === 'function') {
        cb = extension;
        extension = undefined;
      }

      if (!key) {
        return cb(new Error('Invalid key'));
      }

      var path = multihashToPath(key, extension);
      store.remove(path, cb);
    }
  };
};