BlockStorage/repertory-js
1
0
Fork 0
Code Issues Pull Requests Projects Releases Activity

added directory snapshot

Browse Source
This commit is contained in:
sgraves76
2021-03-09 16:05:24 -06:00
parent b553984808
commit e783d490a7
4 changed files with 354 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
README.md
View File

@@ -102,6 +102,19 @@ await api.directory.list('/', async (remote_path, page_count, get_page) => {
} }
}); });
// Asynchronous directory list
const snap = await api.directory.snapshot('/');
try {
for (let i = 0; i < snap.page_count; i++) {
const items = await snap.get_page(i); // Always 'await'
console.log(items);
}
} catch (err) {
console.log(err);
} finally {
await snap.release();
}
// Create new directory // Create new directory
await api.directory.create('/test'); await api.directory.create('/test');

23
src/__tests__/repertory.test.js
View File

@@ -91,12 +91,12 @@ test('can create and remove a directory using api', async () => {
await conn.disconnect(); await conn.disconnect();
}); });
test('can get directory list using api', async () => { test('can get directory list and snapshot using api', async () => {
const conn = const conn =
await repertory.create_pool(2, TEST_HOST, TEST_PORT, TEST_PASSWORD); await repertory.create_pool(2, TEST_HOST, TEST_PORT, TEST_PASSWORD);
const api = repertory.create_api(conn); const api = repertory.create_api(conn);
await api.directory.list('/', async (remote_path, page_count, get_page) => {
console.log(remote_path, page_count, get_page); const test_results = async (remote_path, page_count, get_page) => {
expect(remote_path).toEqual('/'); expect(remote_path).toEqual('/');
expect(page_count).toBeGreaterThanOrEqual(1); expect(page_count).toBeGreaterThanOrEqual(1);
expect(get_page).toBeInstanceOf(Function); expect(get_page).toBeInstanceOf(Function);
@@ -110,8 +110,23 @@ test('can get directory list using api', async () => {
expect(items[1].directory).toBeTruthy(); expect(items[1].directory).toBeTruthy();
expect(items[1].path).toEqual('..'); expect(items[1].path).toEqual('..');
} }
};
await api.directory.list('/', async (remote_path, page_count, get_page) => {
console.log(remote_path, page_count, get_page);
await test_results(remote_path, page_count, get_page);
}); });
const snap = await api.directory.snapshot('/');
try {
console.log(snap.remote_path, snap.page_count, snap.get_page);
await test_results(snap.remote_path, snap.page_count, snap.get_page);
} catch (err) {
console.log(err);
} finally {
await snap.release();
}
await conn.disconnect(); await conn.disconnect();
}); });

11
src/index.js
View File

@@ -12,10 +12,13 @@ export const connect = async (host_or_ip, port, password) => {
export const create_api = conn => { export const create_api = conn => {
return { return {
directory : { directory : {
list : async (remote_path, page_reader_cb) => create: async remote_path => ops.create_directory(conn, remote_path),
ops.list_directory(conn, remote_path, page_reader_cb), list: async (remote_path, page_reader_cb) =>
create : async remote_path => ops.create_directory(conn, remote_path), ops.list_directory(conn, remote_path, page_reader_cb),
remove : async remote_path => ops.remove_directory(conn, remote_path), remove: async remote_path => ops.remove_directory(conn, remote_path),
snapshot: async remote_path => {
return ops.snapshot_directory(conn, remote_path);
},
}, },
file : { file : {
create_or_open : async remote_path => new file( create_or_open : async remote_path => new file(

602
src/ops/index.js
View File

@@ -4,23 +4,86 @@ import {Uint64BE} from 'int64-buffer';
import file from '../io/file'; import file from '../io/file';
import packet from '../networking/packet'; import packet from '../networking/packet';
export const close_file = const _snapshot_directory = async (conn, remote_path) => {
async (conn, remote_path, handle, optional_thread_id) => {
try { try {
const request = new packet(); const request = new packet();
request.encode_utf8(remote_path); request.encode_utf8(remote_path);
request.encode_ui64(handle);
const response = const response =
await conn.send('::RemoteFUSERelease', request, optional_thread_id); await conn.send('::RemoteJSONCreateDirectorySnapshot', request);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
return response.decode_i32(); const result = response.decode_i32();
if (result === 0) {
const data = JSON.parse(response.decode_utf8());
let released = false;
const release = async () => {
if (!released) {
released = true;
const request = new packet();
request.encode_ui64(data.handle);
await conn.send('::RemoteJSONReleaseDirectorySnapshot', request);
}
};
try {
const get_page = async page => {
try {
const request = new packet();
request.encode_utf8(remote_path);
request.encode_ui64(data.handle);
request.encode_ui32(page);
const response =
await conn.send('::RemoteJSONReadDirectorySnapshot', request);
response.decode_ui32(); // Service flags
const result = response.decode_i32();
if (result === 0 || result === -120) {
const data = JSON.parse(response.decode_utf8());
return data.directory_list;
}
} catch (err) {
await release();
return Promise.reject(new Error(`'get_page' failed: ${err}`));
}
return [];
};
return {
get_page,
page_count: data.page_count,
release,
remote_path,
};
} catch (err) {
await release();
return Promise.reject(new Error(`'snapshot_directory' failed: ${err}`));
}
}
} catch (err) { } catch (err) {
return Promise.reject(new Error(`'close_file' failed: ${err}`)); return Promise.reject(new Error(`'snapshot_directory' failed: ${err}`));
} }
}; };
export const close_file =
async (conn, remote_path, handle, optional_thread_id) => {
try {
const request = new packet();
request.encode_utf8(remote_path);
request.encode_ui64(handle);
const response =
await conn.send('::RemoteFUSERelease', request, optional_thread_id);
response.decode_ui32(); // Service flags
return response.decode_i32();
} catch (err) {
return Promise.reject(new Error(`'close_file' failed: ${err}`));
}
};
export const create_directory = async (conn, remote_path) => { export const create_directory = async (conn, remote_path) => {
try { try {
const request = new packet(); const request = new packet();
@@ -37,27 +100,27 @@ export const create_directory = async (conn, remote_path) => {
}; };
export const create_or_open_file = export const create_or_open_file =
async (conn, remote_path, optional_thread_id) => { async (conn, remote_path, optional_thread_id) => {
try { try {
const request = new packet(); const request = new packet();
request.encode_utf8(remote_path); request.encode_utf8(remote_path);
request.encode_ui16((7 << 6) | (5 << 3)); request.encode_ui16((7 << 6) | (5 << 3));
request.encode_ui32(2 | 4); // Read-Write, Create request.encode_ui32(2 | 4); // Read-Write, Create
const response = const response =
await conn.send('::RemoteFUSECreate', request, optional_thread_id); await conn.send('::RemoteFUSECreate', request, optional_thread_id);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
const result = response.decode_i32(); const result = response.decode_i32();
if (result === 0) { if (result === 0) {
return response.decode_ui64(); return response.decode_ui64();
}
return Promise.reject(new Error(`'create_or_open_file' error: ${result}`));
} catch (err) {
return Promise.reject(new Error(`'create_or_open_file' failed: ${err}`));
} }
};
return Promise.reject(new Error(`'create_or_open_file' error: ${result}`));
} catch (err) {
return Promise.reject(new Error(`'create_or_open_file' failed: ${err}`));
}
};
export const delete_file = async (conn, remote_path) => { export const delete_file = async (conn, remote_path) => {
try { try {
@@ -74,96 +137,96 @@ export const delete_file = async (conn, remote_path) => {
}; };
export const download_file = export const download_file =
async (conn, remote_path, local_path, progress_cb, overwrite, resume) => { async (conn, remote_path, local_path, progress_cb, overwrite, resume) => {
try {
const src = new file(conn, await open_file(conn, remote_path), remote_path);
const cleanup = async fd => {
try {
await src.close();
} catch (err) {
console.log(err);
}
try {
if (fd !== undefined) {
fs.closeSync(fd);
}
} catch (err) {
console.log(err);
}
};
try { try {
const src_size = await src.get_size(); const src = new file(conn, await open_file(conn, remote_path), remote_path);
let dst_fd; const cleanup = async fd => {
try {
await src.close();
} catch (err) {
console.log(err);
}
try {
if (fd !== undefined) {
fs.closeSync(fd);
}
} catch (err) {
console.log(err);
}
};
try { try {
let offset = 0; const src_size = await src.get_size();
if (overwrite) { let dst_fd;
dst_fd = fs.openSync(local_path, 'w+');
} else if (resume) {
dst_fd = fs.openSync(local_path, 'r+');
const dst_size = fs.fstatSync(dst_fd).size; try {
if (dst_size === src_size) { let offset = 0;
await cleanup(dst_fd); if (overwrite) {
return true; dst_fd = fs.openSync(local_path, 'w+');
} } else if (resume) {
dst_fd = fs.openSync(local_path, 'r+');
if (dst_size > src_size) { const dst_size = fs.fstatSync(dst_fd).size;
await cleanup(dst_fd); if (dst_size === src_size) {
return Promise.reject(new Error( await cleanup(dst_fd);
return true;
}
if (dst_size > src_size) {
await cleanup(dst_fd);
return Promise.reject(new Error(
`'download_file' failed: destination is larger than source`)); `'download_file' failed: destination is larger than source`));
} }
offset = dst_size; offset = dst_size;
} else { } else {
if (fs.existsSync(local_path)) { if (fs.existsSync(local_path)) {
await cleanup(dst_fd); await cleanup(dst_fd);
return Promise.reject( return Promise.reject(
new Error(`'download_file' failed: file exists`)); new Error(`'download_file' failed: file exists`));
}
dst_fd = fs.openSync(local_path, 'wx+');
} }
dst_fd = fs.openSync(local_path, 'wx+'); let remain = src_size - offset;
} while (remain > 0) {
const to_write = remain >= 65536 ? 65536 : remain;
let remain = src_size - offset; const buffer = await src.read(offset, to_write);
while (remain > 0) { const written = fs.writeSync(dst_fd, buffer, 0, to_write, offset);
const to_write = remain >= 65536 ? 65536 : remain; if (written > 0) {
const buffer = await src.read(offset, to_write); remain -= written;
const written = fs.writeSync(dst_fd, buffer, 0, to_write, offset); offset += written;
if (written > 0) { if (progress_cb) {
remain -= written; progress_cb(local_path, remote_path,
offset += written; ((src_size - remain) / src_size) * 100.0, false);
if (progress_cb) { }
progress_cb(local_path, remote_path,
((src_size - remain) / src_size) * 100.0, false);
} }
} }
}
if (progress_cb) { if (progress_cb) {
progress_cb(local_path, remote_path, 100, true); progress_cb(local_path, remote_path, 100, true);
} }
await cleanup(dst_fd); await cleanup(dst_fd);
return true; return true;
} catch (err) {
await cleanup(dst_fd);
return Promise.reject(new Error(`'download_file' failed: ${err}`));
}
} catch (err) { } catch (err) {
await cleanup(dst_fd); await cleanup();
return Promise.reject(new Error(`'download_file' failed: ${err}`)); return Promise.reject(new Error(`'download_file' failed: ${err}`));
} }
} catch (err) { } catch (err) {
await cleanup();
return Promise.reject(new Error(`'download_file' failed: ${err}`)); return Promise.reject(new Error(`'download_file' failed: ${err}`));
} }
} catch (err) { };
return Promise.reject(new Error(`'download_file' failed: ${err}`));
}
};
export const get_drive_information = async conn => { export const get_drive_information = async conn => {
try { try {
const response = const response =
await conn.send('::RemoteWinFSPGetVolumeInfo', new packet()); await conn.send('::RemoteWinFSPGetVolumeInfo', new packet());
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
const result = response.decode_i32(); const result = response.decode_i32();
@@ -173,86 +236,49 @@ export const get_drive_information = async conn => {
return { return {
free, free,
total, total,
used : (new Uint64BE(total) - new Uint64BE(free)).toString(10), used: (new Uint64BE(total) - new Uint64BE(free)).toString(10),
}; };
} }
return Promise.reject( return Promise.reject(
new Error(`'get_drive_information' failed: ${result}`)); new Error(`'get_drive_information' failed: ${result}`));
} catch (err) { } catch (err) {
return Promise.reject(new Error(`'get_drive_information' failed: ${err}`)); return Promise.reject(new Error(`'get_drive_information' failed: ${err}`));
} }
}; };
export const get_file_attributes = export const get_file_attributes =
async (conn, handle, remote_path, optional_thread_id) => { async (conn, handle, remote_path, optional_thread_id) => {
try { try {
const request = new packet(); const request = new packet();
request.encode_utf8(remote_path); request.encode_utf8(remote_path);
request.encode_ui64(handle); request.encode_ui64(handle);
request.encode_ui32(0); request.encode_ui32(0);
request.encode_ui32(0); request.encode_ui32(0);
const response = const response =
await conn.send('::RemoteFUSEFgetattr', request, optional_thread_id); await conn.send('::RemoteFUSEFgetattr', request, optional_thread_id);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
const result = response.decode_i32(); const result = response.decode_i32();
if (result === 0) { if (result === 0) {
return response.decode_stat(); return response.decode_stat();
}
return Promise.reject(new Error(`'get_file_attributes' failed: ${result}`));
} catch (err) {
return Promise.reject(new Error(`'get_file_attributes' failed: ${err}`));
} }
};
return Promise.reject(new Error(`'get_file_attributes' failed: ${result}`));
} catch (err) {
return Promise.reject(new Error(`'get_file_attributes' failed: ${err}`));
}
};
export const list_directory = async (conn, remote_path, page_reader_cb) => { export const list_directory = async (conn, remote_path, page_reader_cb) => {
const dir_snapshot = await _snapshot_directory(conn, remote_path);
try { try {
const request = new packet(); await page_reader_cb(dir_snapshot.remote_path, dir_snapshot.page_count, dir_snapshot.get_page);
request.encode_utf8(remote_path); await dir_snapshot.release();
const response =
await conn.send('::RemoteJSONCreateDirectorySnapshot', request);
response.decode_ui32(); // Service flags
const result = response.decode_i32();
if (result === 0) {
const data = JSON.parse(response.decode_utf8());
const cleanup = async () => {
const request = new packet();
request.encode_ui64(data.handle);
await conn.send('::RemoteJSONReleaseDirectorySnapshot', request);
};
try {
const get_page = async page => {
const request = new packet();
request.encode_utf8(remote_path);
request.encode_ui64(data.handle);
request.encode_ui32(page);
const response =
await conn.send('::RemoteJSONReadDirectorySnapshot', request);
response.decode_ui32(); // Service flags
const result = response.decode_i32();
if (result === 0 || result === -120) {
const data = JSON.parse(response.decode_utf8());
return data.directory_list;
}
return [];
};
await page_reader_cb(remote_path, data.page_count, get_page);
await cleanup();
} catch (err) {
await cleanup();
return Promise.reject(new Error(`'list_directory' failed: ${err}`));
}
}
} catch (err) { } catch (err) {
return Promise.reject(new Error(`'list_directory' failed: ${err}`)); await dir_snapshot.release();
return Promise.reject(`'list_directory' failed: ${err}`);
} }
}; };
@@ -263,7 +289,7 @@ export const open_file = async (conn, remote_path, optional_thread_id) => {
request.encode_ui32(2); // Read-Write request.encode_ui32(2); // Read-Write
const response = const response =
await conn.send('::RemoteFUSEOpen', request, optional_thread_id); await conn.send('::RemoteFUSEOpen', request, optional_thread_id);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
const result = response.decode_i32(); const result = response.decode_i32();
@@ -277,27 +303,27 @@ export const open_file = async (conn, remote_path, optional_thread_id) => {
}; };
export const read_file = export const read_file =
async (conn, handle, remote_path, offset, length, optional_thread_id) => { async (conn, handle, remote_path, offset, length, optional_thread_id) => {
try { try {
const request = new packet(); const request = new packet();
request.encode_utf8(remote_path); request.encode_utf8(remote_path);
request.encode_ui64(length); request.encode_ui64(length);
request.encode_ui64(offset); request.encode_ui64(offset);
request.encode_ui64(handle); request.encode_ui64(handle);
const response = const response =
await conn.send('::RemoteFUSERead', request, optional_thread_id); await conn.send('::RemoteFUSERead', request, optional_thread_id);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
const result = response.decode_i32(); const result = response.decode_i32();
if (result === length) { if (result === length) {
return response.decode_buffer(result); return response.decode_buffer(result);
}
return Promise.reject(new Error(`'read_file' error: ${result}`));
} catch (err) {
return Promise.reject(new Error(`'read_file' failed: ${err}`));
} }
return Promise.reject(new Error(`'read_file' error: ${result}`)); };
} catch (err) {
return Promise.reject(new Error(`'read_file' failed: ${err}`));
}
};
export const remove_directory = async (conn, remote_path) => { export const remove_directory = async (conn, remote_path) => {
try { try {
@@ -313,146 +339,148 @@ export const remove_directory = async (conn, remote_path) => {
} }
}; };
export const snapshot_directory = _snapshot_directory;
export const truncate_file = export const truncate_file =
async (conn, handle, remote_path, length, optional_thread_id) => { async (conn, handle, remote_path, length, optional_thread_id) => {
try { try {
const request = new packet(); const request = new packet();
request.encode_utf8(remote_path); request.encode_utf8(remote_path);
request.encode_ui64(length); request.encode_ui64(length);
request.encode_ui64(handle); request.encode_ui64(handle);
const response = const response =
await conn.send('::RemoteFUSEFtruncate', request, optional_thread_id); await conn.send('::RemoteFUSEFtruncate', request, optional_thread_id);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
return response.decode_i32(); return response.decode_i32();
} catch (err) { } catch (err) {
return Promise.reject(new Error(`'truncate_file' failed: ${err}`)); return Promise.reject(new Error(`'truncate_file' failed: ${err}`));
} }
}; };
export const upload_file = export const upload_file =
async (conn, local_path, remote_path, progress_cb, overwrite, resume) => { async (conn, local_path, remote_path, progress_cb, overwrite, resume) => {
try {
const src_fd = fs.openSync(local_path, 'r');
const cleanup = async f => {
try {
fs.closeSync(src_fd);
} catch (err) {
console.log(err);
}
try {
if (f) {
await f.close();
}
} catch (err) {
console.log(err);
}
};
try { try {
const src_st = fs.fstatSync(src_fd); const src_fd = fs.openSync(local_path, 'r');
let dst; const cleanup = async f => {
const create_dest = async () => { try {
dst = new file(conn, await create_or_open_file(conn, remote_path), fs.closeSync(src_fd);
remote_path); } catch (err) {
}; console.log(err);
try {
let offset = 0;
if (overwrite) {
await create_dest();
const result = await dst.truncate(0);
if (result !== 0) {
await cleanup(dst);
return Promise.reject(new Error(`'upload_file' failed: ${result}`));
}
} else if (resume) {
await create_dest();
const dst_size = new Uint64BE(await dst.get_size()).toNumber();
if (dst_size === src_st.size) {
await cleanup(dst);
return true;
}
if (dst_size > src_st.size) {
await cleanup(dst);
return Promise.reject(new Error(
`'upload_file' failed: destination is larger than source`));
}
offset = dst_size;
} else {
try {
const f =
new file(conn, await open_file(conn, remote_path), remote_path);
await cleanup(f);
return Promise.reject(
new Error("'upload_file' failed: file exists"));
} catch (err) {
await create_dest();
}
} }
try {
if (f) {
await f.close();
}
} catch (err) {
console.log(err);
}
};
try {
const src_st = fs.fstatSync(src_fd);
let dst;
const create_dest = async () => {
dst = new file(conn, await create_or_open_file(conn, remote_path),
remote_path);
};
let remain = src_st.size - offset; try {
const default_buffer = Buffer.alloc(65536 * 2); let offset = 0;
while (remain > 0) { if (overwrite) {
const to_write = await create_dest();
remain >= default_buffer.length ? default_buffer.length : remain; const result = await dst.truncate(0);
const buffer = to_write === default_buffer.length if (result !== 0) {
? default_buffer await cleanup(dst);
: Buffer.alloc(to_write); return Promise.reject(new Error(`'upload_file' failed: ${result}`));
fs.readSync(src_fd, buffer, 0, to_write, offset); }
const written = await dst.write(offset, buffer); } else if (resume) {
if (written > 0) { await create_dest();
remain -= written; const dst_size = new Uint64BE(await dst.get_size()).toNumber();
offset += written; if (dst_size === src_st.size) {
if (progress_cb) { await cleanup(dst);
progress_cb(local_path, remote_path, return true;
((src_st.size - remain) / src_st.size) * 100.0, }
false);
if (dst_size > src_st.size) {
await cleanup(dst);
return Promise.reject(new Error(
`'upload_file' failed: destination is larger than source`));
}
offset = dst_size;
} else {
try {
const f =
new file(conn, await open_file(conn, remote_path), remote_path);
await cleanup(f);
return Promise.reject(
new Error('\'upload_file\' failed: file exists'));
} catch (err) {
await create_dest();
} }
} }
}
if (progress_cb) { let remain = src_st.size - offset;
progress_cb(local_path, remote_path, 100, true); const default_buffer = Buffer.alloc(65536 * 2);
} while (remain > 0) {
const to_write =
remain >= default_buffer.length ? default_buffer.length : remain;
const buffer = to_write === default_buffer.length
? default_buffer
: Buffer.alloc(to_write);
fs.readSync(src_fd, buffer, 0, to_write, offset);
const written = await dst.write(offset, buffer);
if (written > 0) {
remain -= written;
offset += written;
if (progress_cb) {
progress_cb(local_path, remote_path,
((src_st.size - remain) / src_st.size) * 100.0,
false);
}
}
}
await cleanup(dst); if (progress_cb) {
return true; progress_cb(local_path, remote_path, 100, true);
}
await cleanup(dst);
return true;
} catch (err) {
await cleanup(dst);
return Promise.reject(new Error(`'upload_file' failed: ${err}`));
}
} catch (err) { } catch (err) {
await cleanup(dst); await cleanup();
return Promise.reject(new Error(`'upload_file' failed: ${err}`)); return Promise.reject(new Error(`'upload_file' failed: ${err}`));
} }
} catch (err) { } catch (err) {
await cleanup();
return Promise.reject(new Error(`'upload_file' failed: ${err}`)); return Promise.reject(new Error(`'upload_file' failed: ${err}`));
} }
} catch (err) { };
return Promise.reject(new Error(`'upload_file' failed: ${err}`));
}
};
export const write_file = export const write_file =
async (conn, handle, remote_path, offset, buffer, optional_thread_id) => { async (conn, handle, remote_path, offset, buffer, optional_thread_id) => {
try { try {
const request = new packet(); const request = new packet();
request.encode_utf8(remote_path); request.encode_utf8(remote_path);
request.encode_ui64(buffer.length); request.encode_ui64(buffer.length);
request.encode_buffer(buffer); request.encode_buffer(buffer);
request.encode_ui64(offset); request.encode_ui64(offset);
request.encode_ui64(handle); request.encode_ui64(handle);
const response = const response =
await conn.send('::RemoteFUSEWrite', request, optional_thread_id); await conn.send('::RemoteFUSEWrite', request, optional_thread_id);
response.decode_ui32(); // Service flags response.decode_ui32(); // Service flags
const result = response.decode_i32(); const result = response.decode_i32();
if (result === buffer.length) { if (result === buffer.length) {
return result; return result;
}
return Promise.reject(new Error(`'write_file' error: ${result}`));
} catch (err) {
return Promise.reject(new Error(`'write_file' failed: ${err}`));
} }
return Promise.reject(new Error(`'write_file' error: ${result}`)); };
} catch (err) {
return Promise.reject(new Error(`'write_file' failed: ${err}`));
}
};