You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by su...@apache.org on 2023/11/18 16:34:57 UTC
(incubator-opendal) branch nodejs-stream created (now e9003ee57)
This is an automated email from the ASF dual-hosted git repository.
suyanhanx pushed a change to branch nodejs-stream
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
at e9003ee57 stream init
This branch includes the following new commits:
new e9003ee57 stream init
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-opendal) 01/01: stream init
Posted by su...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
suyanhanx pushed a commit to branch nodejs-stream
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit e9003ee57db86df557ec469688c8b8d499403fd5
Author: suyanhanx <su...@gmail.com>
AuthorDate: Sun Nov 19 00:34:37 2023 +0800
stream init
Signed-off-by: suyanhanx <su...@gmail.com>
---
bindings/nodejs/generated.js | 4 +-
bindings/nodejs/index.d.ts | 72 ++++++++++++++++++
bindings/nodejs/index.js | 24 ++++++
bindings/nodejs/src/lib.rs | 106 +++++++++++++++++++++++++++
bindings/nodejs/tests/fixtures/random.txt | 1 +
bindings/nodejs/tests/suites/async.suite.mjs | 14 +++-
bindings/nodejs/tests/suites/sync.suite.mjs | 17 ++++-
7 files changed, 233 insertions(+), 5 deletions(-)
diff --git a/bindings/nodejs/generated.js b/bindings/nodejs/generated.js
index 2b40ec25e..357723172 100644
--- a/bindings/nodejs/generated.js
+++ b/bindings/nodejs/generated.js
@@ -271,11 +271,13 @@ if (!nativeBinding) {
throw new Error(`Failed to load native binding`)
}
-const { Operator, Entry, Metadata, Lister, BlockingLister, Layer, RetryLayer } = nativeBinding
+const { Operator, Entry, Metadata, BlockingWriter, Writer, Lister, BlockingLister, Layer, RetryLayer } = nativeBinding
module.exports.Operator = Operator
module.exports.Entry = Entry
module.exports.Metadata = Metadata
+module.exports.BlockingWriter = BlockingWriter
+module.exports.Writer = Writer
module.exports.Lister = Lister
module.exports.BlockingLister = BlockingLister
module.exports.Layer = Layer
diff --git a/bindings/nodejs/index.d.ts b/bindings/nodejs/index.d.ts
index 6c6ceba31..6869bea54 100644
--- a/bindings/nodejs/index.d.ts
+++ b/bindings/nodejs/index.d.ts
@@ -147,6 +147,8 @@ export class Operator {
* ```
*/
write(path: string, content: Buffer | string): Promise<void>
+ writer(path: string): Promise<Writer>
+ writerSync(path: string): BlockingWriter
/**
* Write bytes into path synchronously.
*
@@ -429,6 +431,76 @@ export class Metadata {
*/
get lastModified(): string | null
}
+export class BlockingWriter {
+ /**
+ * # Safety
+ *
+ * > &mut self in async napi methods should be marked as unsafe
+ *
+ * Write bytes into this writer.
+ *
+ * ### Example
+ * ```javascript
+ * const writer = await op.writer("path/to/file");
+ * await writer.write(Buffer.from("hello world"));
+ * await writer.close();
+ * ```
+ */
+ write(content: Buffer | string): void
+ /**
+ * # Safety
+ *
+ * > &mut self in async napi methods should be marked as unsafe
+ *
+ * Close this writer.
+ *
+ * ### Example
+ *
+ * ```javascript
+ * const writer = op.writerSync("path/to/file");
+ * writer.write(Buffer.from("hello world"));
+ * writer.close();
+ * ```
+ */
+ close(): void
+}
+/**
+ * Writer
+ *
+ * Could be used to write bytes into a file.
+ */
+export class Writer {
+ /**
+ * # Safety
+ *
+ * > &mut self in async napi methods should be marked as unsafe
+ *
+ * Write bytes into this writer.
+ *
+ * ### Example
+ * ```javascript
+ * const writer = await op.writer("path/to/file");
+ * await writer.write(Buffer.from("hello world"));
+ * await writer.close();
+ * ```
+ */
+ write(content: Buffer | string): Promise<void>
+ /**
+ * # Safety
+ *
+ * > &mut self in async napi methods should be marked as unsafe
+ *
+ * Close this writer.
+ *
+ * ### Example
+ * ```javascript
+ * const writer = await op.writer("path/to/file");
+ * await writer.write(Buffer.from("hello world"));
+ * await writer.close();
+ * ```
+ */
+ close(): Promise<void>
+}
export class Lister {
/**
* # Safety
diff --git a/bindings/nodejs/index.js b/bindings/nodejs/index.js
index c1af66747..c60016a08 100644
--- a/bindings/nodejs/index.js
+++ b/bindings/nodejs/index.js
@@ -19,9 +19,33 @@
/// <reference types="node" />
+const { Writable } = require('node:stream')
+
+class WriterStream extends Writable {
+ constructor(operator, path, options) {
+ super(options)
+ this.writer = operator.writerSync(path)
+ }
+
+ _write(chunk, encoding, callback) {
+ this.writer.write(chunk)
+ if (typeof callback === 'function') {
+ callback()
+ }
+ }
+
+ _final(callback) {
+ this.writer.close()
+ if (typeof callback === 'function') {
+ callback()
+ }
+ }
+}
+
const { Operator, RetryLayer } = require('./generated.js')
module.exports.Operator = Operator
module.exports.layers = {
RetryLayer,
}
+module.exports.WriterStream = WriterStream
diff --git a/bindings/nodejs/src/lib.rs b/bindings/nodejs/src/lib.rs
index 20e46d7e2..bf9fcac0d 100644
--- a/bindings/nodejs/src/lib.rs
+++ b/bindings/nodejs/src/lib.rs
@@ -197,6 +197,18 @@ impl Operator {
self.0.write(&path, c).await.map_err(format_napi_error)
}
+ #[napi]
+ pub async fn writer(&self, path: String) -> Result<Writer> {
+ let w = self.0.writer(&path).await.map_err(format_napi_error)?;
+ Ok(Writer(w))
+ }
+
+ #[napi]
+ pub fn writer_sync(&self, path: String) -> Result<BlockingWriter>{
+ let w = self.0.blocking().writer(&path).map_err(format_napi_error)?;
+ Ok(BlockingWriter(w))
+ }
+
/// Write bytes into path synchronously.
///
/// ### Example
@@ -608,6 +620,100 @@ impl Metadata {
}
}
+#[napi]
+pub struct BlockingWriter(opendal::BlockingWriter);
+
+#[napi]
+impl BlockingWriter{
+ /// # Safety
+ ///
+ /// > &mut self in async napi methods should be marked as unsafe
+ ///
+ /// Write bytes into this writer.
+ ///
+ /// ### Example
+ /// ```javascript
+ /// const writer = await op.writer("path/to/file");
+ /// await writer.write(Buffer.from("hello world"));
+ /// await writer.close();
+ /// ```
+ #[napi]
+ pub unsafe fn write(&mut self, content: Either<Buffer, String>) -> Result<()> {
+ let c = match content {
+ Either::A(buf) => buf.as_ref().to_owned(),
+ Either::B(s) => s.into_bytes(),
+ };
+ self.0.write(c).map_err(format_napi_error)
+ }
+
+ /// # Safety
+ ///
+ /// > &mut self in async napi methods should be marked as unsafe
+ ///
+ /// Close this writer.
+ ///
+ /// ### Example
+ ///
+ /// ```javascript
+ /// const writer = op.writerSync("path/to/file");
+ /// writer.write(Buffer.from("hello world"));
+ /// writer.close();
+ /// ```
+ #[napi]
+ pub unsafe fn close(&mut self) -> Result<()> {
+ self.0.close().map_err(format_napi_error)
+ }
+}
+
+/// Writer
+///
+/// Could be used to write bytes into a file.
+#[napi]
+pub struct Writer(opendal::Writer);
+
+#[napi]
+impl Writer {
+ /// # Safety
+ ///
+ /// > &mut self in async napi methods should be marked as unsafe
+ ///
+ /// Write bytes into this writer.
+ ///
+ /// ### Example
+ /// ```javascript
+ /// const writer = await op.writer("path/to/file");
+ /// await writer.write(Buffer.from("hello world"));
+ /// await writer.close();
+ /// ```
+ #[napi]
+ pub async unsafe fn write(&mut self, content: Either<Buffer, String>) -> Result<()> {
+ let c = match content {
+ Either::A(buf) => buf.as_ref().to_owned(),
+ Either::B(s) => s.into_bytes(),
+ };
+ self.0.write(c).await.map_err(format_napi_error)
+ }
+
+
+
+ /// # Safety
+ ///
+ /// > &mut self in async napi methods should be marked as unsafe
+ ///
+ /// Close this writer.
+ ///
+ /// ### Example
+ /// ```javascript
+ /// const writer = await op.writer("path/to/file");
+ /// await writer.write(Buffer.from("hello world"));
+ /// await writer.close();
+ /// ```
+ #[napi]
+ pub async unsafe fn close(&mut self) -> Result<()> {
+ self.0.close().await.map_err(format_napi_error)
+ }
+}
+
#[napi]
pub struct Lister(opendal::Lister);
diff --git a/bindings/nodejs/tests/fixtures/random.txt b/bindings/nodejs/tests/fixtures/random.txt
new file mode 100644
index 000000000..9896b0e28
--- /dev/null
+++ b/bindings/nodejs/tests/fixtures/random.txt
@@ -0,0 +1 @@
+opendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalopendalop [...]
diff --git a/bindings/nodejs/tests/suites/async.suite.mjs b/bindings/nodejs/tests/suites/async.suite.mjs
index 9a68f804e..0931121e1 100644
--- a/bindings/nodejs/tests/suites/async.suite.mjs
+++ b/bindings/nodejs/tests/suites/async.suite.mjs
@@ -16,9 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
-
import { randomUUID } from 'node:crypto'
-import { expect, test } from 'vitest'
+import { test } from 'vitest'
import { generateBytes } from '../utils.mjs'
export function run(operator) {
@@ -31,4 +30,15 @@ export function run(operator) {
assert.include(error.message, 'NotFound')
}
})
+
+ test('async writer', async () => {
+ const filename = `random_file_${randomUUID()}`
+ const writer = await operator.writer(filename)
+ const data = generateBytes(1024)
+ await writer.write(data)
+ await writer.write(data)
+ await writer.close()
+ const stat = await operator.stat(filename)
+ assert.equal(stat.contentLength, data.length * 2)
+ })
}
diff --git a/bindings/nodejs/tests/suites/sync.suite.mjs b/bindings/nodejs/tests/suites/sync.suite.mjs
index 5e73444d3..732f8e5b2 100644
--- a/bindings/nodejs/tests/suites/sync.suite.mjs
+++ b/bindings/nodejs/tests/suites/sync.suite.mjs
@@ -18,8 +18,10 @@
*/
import { randomUUID } from 'node:crypto'
-import { expect, test } from 'vitest'
-import { generateBytes } from '../utils.mjs'
+import { test } from 'vitest'
+import fs from 'node:fs'
+import path from 'node:path'
+import { WriterStream } from '../../index.js'
export function run(operator) {
test('sync stat not exist files', () => {
@@ -31,4 +33,15 @@ export function run(operator) {
assert.include(error.message, 'NotFound')
}
})
+
+ test('blocking writer stream', async () => {
+ const filename = `random_file_${randomUUID()}`
+ const r = fs.createReadStream(path.resolve(__dirname, '../fixtures/random.txt'))
+ const w = new WriterStream(operator, filename)
+ r.pipe(w)
+ w.on('finish', () => {
+ const t = operator.statSync(filename)
+ assert.equal(t.contentLength, fs.statSync(path.resolve(__dirname, '../fixtures/random.txt')).size)
+ })
+ })
}