-
Notifications
You must be signed in to change notification settings - Fork 53
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
351 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,84 @@ | ||
import { connect, StringCodec } from "../../src/mod.ts"; | ||
|
||
const nc = await connect(); | ||
|
||
const sc = StringCodec(); | ||
const js = nc.jetstream(); | ||
// create the named KV or bind to it if it exists: | ||
const kv = await js.views.kv("testing", { history: 5 }); | ||
|
||
// create an entry - this is similar to a put, but will fail if the | ||
// key exists | ||
await kv.create("hello.world", sc.encode("hi")); | ||
|
||
// Values in KV are stored as KvEntries: | ||
// { | ||
// bucket: string, | ||
// key: string, | ||
// value: Uint8Array, | ||
// created: Date, | ||
// revision: number, | ||
// delta?: number, | ||
// operation: "PUT"|"DEL"|"PURGE" | ||
// } | ||
// The operation property specifies whether the value was | ||
// updated (PUT), deleted (DEL) or purged (PURGE). | ||
|
||
// you can monitor values modification in a KV by watching. | ||
// You can watch specific key subset or everything. | ||
// Watches start with the latest value for each key in the | ||
// set of keys being watched - in this case all keys | ||
const watch = await kv.watch(); | ||
(async () => { | ||
for await (const e of watch) { | ||
// do something with the change | ||
console.log( | ||
`watch: ${e.key}: ${e.operation} ${e.value ? sc.decode(e.value) : ""}`, | ||
); | ||
} | ||
})().then(); | ||
|
||
// update the entry | ||
await kv.put("hello.world", sc.encode("world")); | ||
// retrieve the KvEntry storing the value | ||
// returns null if the value is not found | ||
const e = await kv.get("hello.world"); | ||
// initial value of "hi" was overwritten above | ||
console.log(`value for get ${sc.decode(e!.value)}`); | ||
|
||
const buf: string[] = []; | ||
const keys = await kv.keys(); | ||
await (async () => { | ||
for await (const k of keys) { | ||
buf.push(k); | ||
} | ||
})(); | ||
console.log(`keys contains hello.world: ${buf[0] === "hello.world"}`); | ||
|
||
let h = await kv.history({ key: "hello.world" }); | ||
await (async () => { | ||
for await (const e of h) { | ||
// do something with the historical value | ||
// you can test e.operation for "PUT", "DEL", or "PURGE" | ||
// to know if the entry is a marker for a value set | ||
// or for a deletion or purge. | ||
console.log( | ||
`history: ${e.key}: ${e.operation} ${e.value ? sc.decode(e.value) : ""}`, | ||
); | ||
} | ||
})(); | ||
|
||
// deletes the key - the delete is recorded | ||
await kv.delete("hello.world"); | ||
|
||
// purge is like delete, but all history values | ||
// are dropped and only the purge remains. | ||
await kv.purge("hello.world"); | ||
|
||
// stop the watch operation above | ||
watch.stop(); | ||
|
||
// danger: destroys all values in the KV! | ||
await kv.destroy(); | ||
|
||
await nc.close(); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,123 @@ | ||
import { connect, StorageType, StringCodec } from "../../src/mod.ts"; | ||
|
||
const nc = await connect(); | ||
|
||
const sc = StringCodec(); | ||
const js = nc.jetstream(); | ||
// create the named ObjectStore or bind to it if it exists: | ||
const os = await js.views.os("testing", { storage: StorageType.File }); | ||
|
||
// ReadableStreams allows JavaScript to work with large data without | ||
// necessarily keeping it all in memory. | ||
// | ||
// ObjectStore reads and writes to JetStream via ReadableStreams | ||
// https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream | ||
// You can easily create ReadableStreams from static data or iterators | ||
|
||
// here's an example of creating a readable stream from static data | ||
function readableStreamFrom(data: Uint8Array): ReadableStream<Uint8Array> { | ||
return new ReadableStream<Uint8Array>({ | ||
pull(controller) { | ||
// the readable stream adds data | ||
controller.enqueue(data); | ||
controller.close(); | ||
}, | ||
}); | ||
} | ||
|
||
// reading from a ReadableStream is similar to working with an async iterator: | ||
async function fromReadableStream( | ||
rs: ReadableStream<Uint8Array>, | ||
) { | ||
let i = 1; | ||
const reader = rs.getReader(); | ||
while (true) { | ||
const { done, value } = await reader.read(); | ||
if (done) { | ||
break; | ||
} | ||
if (value && value.length) { | ||
// do something with the accumulated data | ||
console.log(`chunk ${i++}: ${sc.decode(value)}`); | ||
} | ||
} | ||
} | ||
|
||
let e = await os.get("hello"); | ||
console.log(`hello entry exists? ${e !== null}`); | ||
|
||
// watch notifies when a change in the object store happens | ||
const watch = await os.watch(); | ||
(async () => { | ||
for await (const i of watch) { | ||
// when asking for history you get a null | ||
// that tells you when all the existing values | ||
// are provided | ||
if (i === null) { | ||
continue; | ||
} | ||
console.log(`watch: ${i!.name} deleted?: ${i!.deleted}`); | ||
} | ||
})(); | ||
|
||
// putting an object returns an info describing the object | ||
const info = await os.put({ | ||
name: "hello", | ||
description: "first entry", | ||
options: { | ||
max_chunk_size: 1, | ||
}, | ||
}, readableStreamFrom(sc.encode("hello world"))); | ||
|
||
console.log( | ||
`object size: ${info.size} number of chunks: ${info.size} deleted: ${info.deleted}`, | ||
); | ||
|
||
// let's add oen more asset | ||
await os.put({ | ||
name: "hello2", | ||
}, readableStreamFrom(sc.encode("hi there"))); | ||
|
||
// reading it back: | ||
const r = await os.get("hello"); | ||
// it is possible while we read the ReadableStream that something goes wrong | ||
// the error property on the result will resolve to null if there's no error | ||
// otherwise to the error from the ReadableStream | ||
r?.error.then((err) => { | ||
if (err) { | ||
console.error("reading the readable stream failed:", err); | ||
} | ||
}); | ||
|
||
// use our sample stream reader to process output to the console | ||
// chunk 1: h | ||
// chunk 2: e | ||
// ... | ||
// chunk 11: d | ||
await fromReadableStream(r!.data); | ||
|
||
// list all the entries in the object store | ||
// returns the info for each entry | ||
const list = await os.list(); | ||
list.forEach((i) => { | ||
console.log(`list: ${i.name}`); | ||
}); | ||
|
||
// you can also get info on the object store as a whole: | ||
const status = await os.status(); | ||
console.log(`bucket: '${status.bucket}' size in bytes: ${status.size}`); | ||
|
||
// you can delete an entry so long as the objectstore is not sealed | ||
await os.delete("hello2"); | ||
|
||
// you can prevent additional modifications by sealing it | ||
const final = await os.seal(); | ||
console.log(`bucket: '${final.bucket}' sealed: ${final.sealed}`); | ||
|
||
// only other thing that you can do is destroy it | ||
// this gets rid of the object store's underlying stream | ||
// note all data will be gone forever | ||
const destroyed = await os.destroy(); | ||
console.log(`destroyed: ${destroyed}`); | ||
|
||
await nc.close(); |
Oops, something went wrong.