fluencelabs/dweb-transports
1
0
Fork 0
mirror of https://github.com/fluencelabs/dweb-transports synced 2025-03-14 18:10:49 +00:00
Code Issues Projects Releases Wiki Activity

Replace verbose with debug package for logging - looks good

Browse Source
This commit is contained in:
Mitra Ardron 2018-08-13 17:55:04 +10:00
parent 9b3b776733
commit cc03e4334c
15 changed files with 6833 additions and 659 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
API.md
View File

@ -13,12 +13,10 @@ We use a naming convention that anything starting “p_” returns a promise so
Ideally functions should take a String, Buffer or where applicable Object as parameters with automatic conversion.
And anything that takes a URL should take either a string or parsed URL object.
The verbose parameter is a boolean that is an indicator of a need to output to the console. Normally it will be passed down to called functions and default to false.
Note that example_block.html collects this from the URL and passes it to the library,
which is intended to be a good way to see what is happening.
Note: I am gradually (March2018) changing the API to take an opts {} dict which includes verbose as one field. This process is incomplete, but I’m happy to see it accelerated if there is any code built on this, just let mitra@archive.org know.
Note: I am gradually (March2018) changing the API to take an opts {} dict. This process is incomplete, but I’m happy to see it accelerated if there is any code built on this, just let mitra@archive.org know.
## Overview
@ -55,19 +53,18 @@ status|Numeric indication of transport status: Started(0); Failed(1); Starting(2
Transport setup is split into 3 parts, this allows the Transports class to do the first phase on all the transports synchronously,
then asynchronously (or at a later point) try and connect to all of them in parallel.
##### static setup0 (options, verbose)
##### static setup0 (options)
First part of setup, create obj, add to Transports but dont attempt to connect, typically called instead of p_setup if want to parallelize connections. In almost all cases this will call the constructor of the subclass
Should be synchronous and leave `status=STATUS_LOADED`
```
options Object fields including those needed by transport layer
verbose boolean - true for debugging output
Resolves to Instance of subclass of Transport
```
Default options should be set in each transport, but can be overwritten,
for example to overwrite the options for HTTP call it with
`options={ http: { urlbase: “https://dweb.me:443/” } }`
##### async p_setup1 (verbose, cb)
##### async p_setup1 (, cb)
Setup the resource and open any P2P connections etc required to be done just once.
Asynchronous and should leave `status=STATUS_STARTING` until it resolves, or `STATUS_FAILED` if fails.
```
@ -75,7 +72,7 @@ cb (t)=>void If set, will be called back as status changes (so could be multip
Resolves to the Transport instance
```
##### async p_setup2 (verbose, cb)
##### async p_setup2 (, cb)
Works like p_setup1 but runs after p_setup1 has completed for all transports.
This allows for example YJS to wait for IPFS to be connected in TransportIPFS.setup1()
and then connect itself using the IPFS object.
@ -84,7 +81,7 @@ cb (t)=>void If set, will be called back as status changes (so could be multip
Resolves to the Transport instance
```
##### async p_setup(options, verbose, cb)
##### async p_setup(options, cb)
A deprecated utility to simply setup0 then p_setup1 then p_setup2 to allow a transport to be started
in one step, normally `Transports.p_setup` should be called instead.
@ -95,7 +92,7 @@ in the paused state it will not be used for transport but, in some cases, will s
cb(transport)=>void a callback called after this is run, may be used for example to change the UI
```
##### async p_status (verbose)
##### async p_status ()
Check the status of the underlying transport. This may update the "status" field from the underlying transport.
```
returns: a numeric code for the status of a transport.
@ -110,15 +107,14 @@ Code|Name|Means
4|STATUS_PAUSED|It was launched, probably connected, but now paused so will be ignored by validfor()
### Transport: General storage and retrieval of objects
##### p_rawstore(data, {verbose})
##### p_rawstore(data)
Store a opaque blob of data onto the decentralised transport.
```
data string|Buffer data to store - no assumptions made to size or content
verbose boolean - True for debugging output
Resolves to url of data stored
```
##### p_rawfetch(url, {timeoutMS, start, end, relay, verbose})
##### p_rawfetch(url, {timeoutMS, start, end, relay})
Fetch some bytes based on a url, no assumption is made about the data in terms of size or structure.
Where required by the underlying transport it should retrieve a number if its "blocks" and concatenate them.
@ -129,13 +125,12 @@ url string url of object being retrieved in form returned by link or p_
timeoutMS Max time to wait on transports that support it (IPFS for fetch)
start,end Inclusive byte range wanted (must be supported, uses a "slice" on output if transport ignores it.
relay If first transport fails, try and retrieve on 2nd, then store on 1st, and so on.
verbose boolean - True for debugging output
Resolves to string The object being fetched, (note currently (April 2018) returned as a string, may refactor to return Buffer)
throws: TransportError if url invalid - note this happens immediately, not as a catch in the promise
```
### Transport: Handling lists
##### p_rawadd(url, sig, {verbose})
##### p_rawadd(url, sig)
Store a new list item, it should be stored so that it can be retrieved either by "signedby" (using p_rawlist)
or by "url" (with p_rawreverse).
@ -148,51 +143,46 @@ sig Signature data structure (see below - contains url, date, signedby, s
urls - array of urls for the object being signed
signature - verifiable signature of date+urls
signedby - url of data structure (typically CommonList) holding public key used for the signature
verbose boolean - True for debugging output
```
##### p_rawlist(url, {verbose})
##### p_rawlist(url)
Fetch all the objects in a list, these are identified by the url of the public key used for signing.
Note this is the 'signedby' parameter of the p_rawadd call, not the 'url' parameter.
List items may have other data (e.g. reference ids of underlying transport)
```
url String with the url that identifies the list
(this is the 'signedby' parameter of the p_rawadd call, not the 'url' parameter
verbose boolean - True for debugging output
Resolves to Array: An array of objects as stored on the list. Each of which is …..
```
Each item of the list is a dict: {"url": url, "date": date, "signature": signature, "signedby": signedby}
##### p_rawreverse (url, {verbose})
##### p_rawreverse (url)
Similar to p_rawlist, but return the list item of all the places where the object url has been listed.
(not supported by most transports)
```
url String with the url that identifies the object put on a list
This is the “url” parameter of p_rawadd
verbose boolean - True for debugging output
Resolves to Array objects as stored on the list (see p_rawlist for format)
```
##### listmonitor (url, cb, {verbose, current})
##### listmonitor (url, cb, { current})
Setup a callback called whenever an item is added to a list, typically it would be called immediately after a p_rawlist to get any more items not returned by p_rawlist.
```
url Identifier of list (as used by p_rawlist and "signedby" parameter of p_rawadd
cb(obj) function(obj) Callback for each new item added to the list
verbose true for debugging output
current true to send existing members as well as new
obj is same format as p_rawlist or p_rawreverse
```
##### async p_newlisturls(cl, {verbose})
##### async p_newlisturls(cl)
Obtain a pair of URLs for a new list. The transport can use information in the cl to generate this or create something random (the former is encouraged since it means repeat tests might not generate new lists). Possession of the publicurl should be sufficient to read the list, the privateurl should be required to read (for some transports they will be identical, and higher layers should check for example that a signature is signed.
```
cl CommonList instance that can be used as a seed for the URL
verbose boolean - True for debugging output
Returns [privateurl, publicurl]
```
### Transport: Support for KeyValueTable
##### async p_newdatabase(pubkey, {verbose}) {
##### async p_newdatabase(pubkey) {
Create a new database based on some existing object
```
pubkey: Something that is, or has a pubkey, by default support Dweb.PublicPrivate, KeyPair
@ -200,7 +190,7 @@ pubkey: Something that is, or has a pubkey, by default support Dweb.PublicPr
returns: {publicurl, privateurl} which may be the same if there is no write authentication
```
##### async p_newtable(pubkey, table, {verbose}) {
##### async p_newtable(pubkey, table) {
Create a new table,
```
pubkey: Is or has a pubkey (see p_newdatabase)
@ -208,7 +198,7 @@ table: String representing the table - unique to the database
returns: {privateurl, publicurl} which may be the same if there is no write authentication
```
##### async p_set(url, keyvalues, value, {verbose})
##### async p_set(url, keyvalues, value)
Set one or more keys in a table.
```
url: URL of the table
@ -216,7 +206,7 @@ keyvalues: String representing a single key OR dictionary of keys
value: String or other object to be stored (its not defined yet what objects should be supported, e.g. any object ?
```
##### async p_get(url, keys, {verbose})
##### async p_get(url, keys)
Get one or more keys from a table
```
url: URL of the table
@ -224,21 +214,21 @@ keys: Array of keys
returns: Dictionary of values found (undefined if not found)
```
##### async p_delete(url, keys, {verbose})
##### async p_delete(url, keys)
Delete one or more keys from a table
```
url: URL of the table
keys: Array of keys
```
##### async p_keys(url, {verbose})
##### async p_keys(url)
Return a list of keys in a table (suitable for iterating through)
```
url: URL of the table
returns: Array of strings
```
##### async p_getall(url, {verbose})
##### async p_getall(url)
Return a dictionary representing the table
```
url: URL of the table
@ -246,7 +236,7 @@ returns: Dictionary of Key:Value pairs, note take care if this could
```
### Transports - other functions
##### static async p_f_createReadStream(url, {wanturl, verbose})
##### static async p_f_createReadStream(url, {wanturl})
Provide a function of the form needed by <VIDEO> tag and renderMedia library etc
```
url Urls of stream
@ -310,22 +300,22 @@ Returns: Array of pairs of url & transport instance [ [ u1, t1], [u1, t2], [u
Async version of validFor for serviceworker and TransportsProxy
##### static http(verbose)
##### static http()
```
returns instance of TransportHTTP if connected
```
##### static ipfs(verbose)
##### static ipfs()
```
returns instance of TransportIPFS if connected
```
##### static webtorrent(verbose)
##### static webtorrent()
```
returns instance of TransportWEBTORRENT if connected
```
##### static gun(verbose)
##### static gun()
```
returns instance of TransportGUN if connected
```
@ -349,7 +339,7 @@ cb(urls) => urls Provide callback function
t: Add a Transport instance to _transports
```
##### static setup0(transports, options, verbose, cb)
##### static setup0(transports, options, cb)
Calls setup0 for each transport based on its short name. Specially handles ‘LOCAL’ as a transport pointing at a local http server (for testing).
```
transports Array of short names of transports e.g. [‘IPFS’,’HTTP’,’GUN’]
@ -358,10 +348,10 @@ cb Callback to be called each time status changes
Returns: Array of transport instances
```
##### static async p_setup1(verbose, cb)
##### static async p_setup1(, cb)
Call p_setup1 on all transports that were created in setup0(). Completes when all setups are complete.
##### static async p_setup2(verbose, cb)
##### static async p_setup2(, cb)
Call p_setup2 on all transports that were created in setup0(). Completes when all setups are complete.
##### static async refreshstatus(t)
@ -370,7 +360,7 @@ Set the class of t.statuselement (if set) to transportstatus0..transportstatus4
t Instance of transport
```
#### static async p_connect({options}, verbose)
#### static async p_connect({options})
Main connection process for a browser based application,
```
options {
@ -403,24 +393,24 @@ For parameters refer to underlying Transport call
Call|Returns|Behavior
---|---|---
static async p_rawstore(data, {verbose})|[urls]|Tries all and combines results
static async p_rawfetch(urls, {timeoutMS, start, end, verbose, relay})|data|See note
static async p_rawlist(urls, {verbose})|[sigs]|Tries all and combines results
static async p_rawadd(urls, sig, {verbose})||Tries on all urls, error if none succeed
static listmonitor(urls, cb, {verbose, current})||Tries on all urls (so note cb may be called multiple times)
static p_newlisturls(cl, {verbose})|[urls]|Tries all and combines results
static async p_rawstore(data)|[urls]|Tries all and combines results
static async p_rawfetch(urls, {timeoutMS, start, end, relay})|data|See note
static async p_rawlist(urls)|[sigs]|Tries all and combines results
static async p_rawadd(urls, sig)||Tries on all urls, error if none succeed
static listmonitor(urls, cb, { current})||Tries on all urls (so note cb may be called multiple times)
static p_newlisturls(cl)|[urls]|Tries all and combines results
static async p_f_createReadStream(urls, options)|f(opts)=>stream|Returns first success
static async p_get(urls, keys, {verbose})|currently (April 2018) returns on first success, TODO - will combine results and relay across transports
static async p_set(urls, keyvalues, value, {verbose})|Tries all, error if none succeed
static async p_delete(urls, keys, {verbose})|Tries all, error if none succeed
static async p_keys(urls, {verbose}|[keys]|currently (April 2018) returns on first success, TODO - will combine results and relay across transports
static async p_getall(urls, {verbose})|dict|currently (April 2018) returns on first success, TODO - will combine results and relay across transports
static async p_newdatabase(pubkey, {verbose})|{privateurls: [urls], publicurls: [urls]}|Tries all and combines results
static async p_newtable(pubkey, table, {verbose})|{privateurls: [urls], publicurls: [urls]}|Tries all and combines results
static async p_connection(urls, verbose)||Tries all parallel
static monitor(urls, cb, {verbose, current})||Tries all sequentially
static async p_get(urls, keys)|currently (April 2018) returns on first success, TODO - will combine results and relay across transports
static async p_set(urls, keyvalues, value)|Tries all, error if none succeed
static async p_delete(urls, keys)|Tries all, error if none succeed
static async p_keys(urls|[keys]|currently (April 2018) returns on first success, TODO - will combine results and relay across transports
static async p_getall(urls)|dict|currently (April 2018) returns on first success, TODO - will combine results and relay across transports
static async p_newdatabase(pubkey)|{privateurls: [urls], publicurls: [urls]}|Tries all and combines results
static async p_newtable(pubkey, table)|{privateurls: [urls], publicurls: [urls]}|Tries all and combines results
static async p_connection(urls)||Tries all parallel
static monitor(urls, cb, { current})||Tries all sequentially
##### static async p_rawfetch(urls, {timeoutMS, start, end, verbose, relay})
##### static async p_rawfetch(urls, {timeoutMS, start, end, relay})
Tries to fetch on all valid transports until successful. See Transport.p_rawfetch
```
timeoutMS: Max time to wait on transports that support it (IPFS for fetch)
@ -432,19 +422,18 @@ relay If first transport fails, try and retrieve on 2nd, then store on 1s
A utility class to support HTTP with or without TransportHTTP
e.g. `httptools.http().p_httpfetch("http://foo.com/bar", {method: 'GET'} )`
##### p_httpfetch(url, init, {verbose)}
##### p_httpfetch(url, init)
Fetch a url.
If the result
url: HTTP or HTTPS url
init: Init parameter to fetch (see for docs)
verbose: boolean for debugging
returns: Depends on mime type;
If application/json returns a Object,
If text/* returns text
Oherwise Buffer
##### p_GET(url, {start, end, verbose})
##### p_GET(url, {start, end})
Shortcut to do a HTTP/POST get, sets `mode: cors, redirect: follow, keepalive: true, cache: default`
start: First byte to retrieve
@ -453,7 +442,7 @@ end: Last byte to retrieve (undefined means end of file)
Note that it passes start and end as the Range header, most servers support it,
but it does not (yet) explicitly check the result.
##### p_POST(url, type, data, verbose)
##### p_POST(url, type, data)
Shortcut to do a HTTP/HTTPS POST. sets same options as p_GET
data: Data to send to fetch, typically the body,

4
README.md
View File

@ -38,7 +38,7 @@ async function main(url) {
statuselement: document.getElementById("statuselement"), // Where to build status indicator
defaulttransports: ["HTTP","IPFS"], // Default transports if not specified
transports: searchparams.getAll("transport") // Allow override default from URL parameters
}, verbose); // Pass verbose global parameter from command line
});
// Any code you want to run after connected to transports goes here.
} catch(err) {
console.log("App Error:", err);
@ -46,8 +46,6 @@ async function main(url) {
}
}
var searchparams = new URL(window.location.href).searchParams;
// Allow specifying ?verbose=true in URL to get debugging, and transport=HTTP etc
var verbose = searchparams.get("verbose") || false;
```
See [example_block.html](./example_block.html) for an example of connecting, storing and retrieving.

108
Transport.js
View File

@ -7,7 +7,7 @@ function delay(ms, val) { return new Promise(resolve => {setTimeout(() => { reso
class Transport {
constructor(options, verbose) {
constructor(options) {
/*
Doesnt do anything, its all done by SuperClasses,
Superclass should merge with default options, call super
@ -19,14 +19,14 @@ class Transport {
*/
}
static setup0(options, verbose) {
static setup0(options) {
/*
First part of setup, create obj, add to Transports but dont attempt to connect, typically called instead of p_setup if want to parallelize connections.
*/
throw new errors.IntentionallyUnimplementedError("Intentionally undefined function Transport.setup0 should have been subclassed");
}
p_setup1(options, verbose) {
p_setup1(options) {
/*
Setup the resource and open any P2P connections etc required to be done just once. Asynchronous and should leave status=STATUS_STARTING until it resolves, or STATUS_FAILED if fails.
@ -35,7 +35,7 @@ class Transport {
*/
return this;
}
p_setup2(options, verbose) {
p_setup2(options) {
/*
Works like p_setup1 but runs after p_setup1 has completed for all transports. This allows for example YJS to wait for IPFS to be connected in TransportIPFS.setup1() and then connect itself using the IPFS object.
@ -44,14 +44,14 @@ class Transport {
*/
return this;
}
static async p_setup(options, verbose, cb) {
static async p_setup(options, cb) {
/*
A deprecated utility to simply setup0 then p_setup1 then p_setup2 to allow a transport to be started in one step, normally Transports.p_setup should be called instead.
*/
let t = await this.setup0(options, verbose) // Sync version that doesnt connect
.p_setup1(verbose, cb); // And connect
let t = await this.setup0(options) // Sync version that doesnt connect
.p_setup1(cb); // And connect
return t.p_setup2(verbose, cb); // And connect
return t.p_setup2(cb); // And connect
}
togglePaused(cb) {
/*
@ -71,7 +71,7 @@ class Transport {
if (cb) cb(this);
}
async p_status(verbose) {
async p_status() {
/*
Check the status of the underlying transport. This may update the "status" field from the underlying transport.
returns: a numeric code for the status of a transport.
@ -104,15 +104,14 @@ class Transport {
Returns a promise that resolves to the url of the data
:param string|Buffer data: Data to store - no assumptions made to size or content
:param boolean verbose: true for debugging output
:resolve string: url of data stored
*/
throw new errors.ToBeImplementedError("Intentionally undefined function Transport.p_rawstore should have been subclassed");
}
async p_rawstoreCaught(data, {verbose}) {
async p_rawstoreCaught(data) {
try {
return await this.p_rawstore(data, {verbose});
return await this.p_rawstore(data);
} catch (err) {
}
@ -123,7 +122,7 @@ class Transport {
//noinspection JSUnusedLocalSymbols
p_rawfetch(url, {timeoutMS=undefined, start=undefined, end=undefined, relay=false, verbose=false}={}) {
p_rawfetch(url, {timeoutMS=undefined, start=undefined, end=undefined, relay=false}={}) {
/*
Fetch some bytes based on a url, no assumption is made about the data in terms of size or structure.
Where required by the underlying transport it should retrieve a number if its "blocks" and concatenate them.
@ -131,7 +130,6 @@ class Transport {
There may also be need for a streaming version of this call, at this point undefined.
:param string url: URL of object being retrieved
:param verbose: true for debugging output
:param timeoutMS Max time to wait on transports that support it (IPFS for fetch)
:param start,end Inclusive byte range wanted (must be supported, uses a "slice" on output if transport ignores it.
:param relay If first transport fails, try and retrieve on 2nd, then store on 1st, and so on.
@ -147,7 +145,7 @@ class Transport {
throw new errors.ToBeImplementedError("Undefined function Transport.p_fetch - may define higher level semantics here (see Python)");
}
p_rawadd(url, sig, {verbose=false}={}) {
p_rawadd(url, sig) {
/*
Store a new list item, ideally it should be stored so that it can be retrieved either by "signedby" (using p_rawlist) or
by "url" (with p_rawreverse). The underlying transport does not need to guarantee the signature,
@ -155,13 +153,12 @@ class Transport {
:param string url: String identifying an object being added to the list.
:param Signature sig: A signature data structure.
:param boolean verbose: true for debugging output
:resolve undefined:
*/
throw new errors.ToBeImplementedError("Undefined function Transport.p_rawadd");
}
p_rawlist(url, {verbose=false}={}) {
p_rawlist(url) {
/*
Fetch all the objects in a list, these are identified by the url of the public key used for signing.
(Note this is the 'signedby' parameter of the p_rawadd call, not the 'url' parameter
@ -170,7 +167,6 @@ class Transport {
List items may have other data (e.g. reference ids of underlying transport)
:param string url: String with the url that identifies the list.
:param boolean verbose: true for debugging output
:resolve array: An array of objects as stored on the list.
*/
throw new errors.ToBeImplementedError("Undefined function Transport.p_rawlist");
@ -179,7 +175,7 @@ class Transport {
p_list() {
throw new Error("Undefined function Transport.p_list");
}
p_newlisturls(cl, {verbose=false}={}) {
p_newlisturls(cl) {
/*
Must be implemented by any list, return a pair of URLS that may be the same, private and public links to the list.
returns: ( privateurl, publicurl) e.g. yjs:xyz/abc or orbitdb:a123
@ -188,27 +184,25 @@ class Transport {
}
//noinspection JSUnusedGlobalSymbols
p_rawreverse(url, {verbose=false}={}) {
p_rawreverse(url) {
/*
Similar to p_rawlist, but return the list item of all the places where the object url has been listed.
The url here corresponds to the "url" parameter of p_rawadd
Returns a promise that resolves to the list.
:param string url: String with the url that identifies the object put on a list.
:param boolean verbose: true for debugging output
:resolve array: An array of objects as stored on the list.
*/
throw new errors.ToBeImplementedError("Undefined function Transport.p_rawreverse");
}
listmonitor(url, callback, {verbose=false, current=false}={}) {
listmonitor(url, callback, {current=false}={}) {
/*
Setup a callback called whenever an item is added to a list, typically it would be called immediately after a p_rawlist to get any more items not returned by p_rawlist.
:param url: string Identifier of list (as used by p_rawlist and "signedby" parameter of p_rawadd
:param callback: function(obj) Callback for each new item added to the list
obj is same format as p_rawlist or p_rawreverse
:param verbose: boolean - true for debugging output
*/
console.log("Undefined function Transport.listmonitor"); // Note intentionally a log, as legitamte to not implement it
}
@ -218,7 +212,7 @@ class Transport {
// Support for Key-Value pairs as per
// https://docs.google.com/document/d/1yfmLRqKPxKwB939wIy9sSaa7GKOzM5PrCZ4W1jRGW6M/edit#
async p_newdatabase(pubkey, {verbose=false}={}) {
async p_newdatabase(pubkey) {
/*
Create a new database based on some existing object
pubkey: Something that is, or has a pubkey, by default support Dweb.PublicPrivate, KeyPair or an array of strings as in the output of keypair.publicexport()
@ -228,7 +222,7 @@ class Transport {
}
//TODO maybe change the listmonitor / monitor code for to use "on" and the structure of PP.events
//TODO but note https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Proxy about Proxy which might be suitable, prob not as doesnt map well to lists
async p_newtable(pubkey, table, {verbose=false}={}) {
async p_newtable(pubkey, table) {
/*
Create a new table,
pubkey: Is or has a pubkey (see p_newdatabase)
@ -238,7 +232,7 @@ class Transport {
throw new errors.ToBeImplementedError("Undefined function Transport.p_newtable");
}
async p_set(url, keyvalues, value, {verbose=false}={}) { // url = yjs:/yjs/database/table/key
async p_set(url, keyvalues, value) { // url = yjs:/yjs/database/table/key
/*
Set one or more keys in a table.
url: URL of the table
@ -247,7 +241,7 @@ class Transport {
*/
throw new errors.ToBeImplementedError("Undefined function Transport.p_set");
}
async p_get(url, keys, {verbose=false}={}) {
async p_get(url, keys) {
/* Get one or more keys from a table
url: URL of the table
keys: Array of keys
@ -256,7 +250,7 @@ class Transport {
throw new errors.ToBeImplementedError("Undefined function Transport.p_get");
}
async p_delete(url, keys, {verbose=false}={}) {
async p_delete(url, keys) {
/* Delete one or more keys from a table
url: URL of the table
keys: Array of keys
@ -264,21 +258,21 @@ class Transport {
throw new errors.ToBeImplementedError("Undefined function Transport.p_delete");
}
async p_keys(url, {verbose=false}={}) {
async p_keys(url) {
/* Return a list of keys in a table (suitable for iterating through)
url: URL of the table
returns: Array of strings
*/
throw new errors.ToBeImplementedError("Undefined function Transport.p_keys");
}
async p_getall(url, {verbose=false}={}) {
async p_getall(url) {
/* Return a dictionary representing the table
url: URL of the table
returns: Dictionary of Key:Value pairs, note take care if this could be large.
*/
throw new errors.ToBeImplementedError("Undefined function Transport.p_keys");
}
static async p_f_createReadStream(url, {wanturl=false, verbose=false}) {
static async p_f_createReadStream(url, {wanturl=false}) {
/*
Provide a function of the form needed by tag and renderMedia library etc
@ -308,59 +302,59 @@ class Transport {
return c;
}
async p_test_list({urlexpectedsubstring=undefined, verbose=false}={}) {
async p_test_list({urlexpectedsubstring=undefined}={}) {
//TODO - this test doesn't work since we dont have Signature nor want to create dependency on it - when works, add to GUN & YJS
if (verbose) {console.log(this.name,"p_test_kvt")}
{console.log(this.name,"p_test_kvt")}
try {
let table = await this.p_newlisturls("NACL VERIFY:1234567LIST", {verbose});
let table = await this.p_newlisturls("NACL VERIFY:1234567LIST");
let mapurl = table.publicurl;
if (verbose) console.log("newlisturls=",mapurl);
console.log("newlisturls=",mapurl);
console.assert((!urlexpectedsubstring) || mapurl.includes(urlexpectedsubstring));
await this.p_rawadd(mapurl, "testvalue", {verbose});
let res = await this.p_rawlist(mapurl, {verbose});
await this.p_rawadd(mapurl, "testvalue");
let res = await this.p_rawlist(mapurl);
console.assert(res.length===1 && res[0] === "testvalue");
await this.p_rawadd(mapurl, {foo: "bar"}, {verbose}); // Try adding an object
res = await this.p_rawlist(mapurl, {verbose});
await this.p_rawadd(mapurl, {foo: "bar"}); // Try adding an object
res = await this.p_rawlist(mapurl);
console.assert(res.length === 2 && res[1].foo === "bar");
await this.p_rawadd(mapurl, [1,2,3], {verbose}); // Try setting to an array
res = await this.p_rawlist(mapurl, {verbose});
await this.p_rawadd(mapurl, [1,2,3]); // Try setting to an array
res = await this.p_rawlist(mapurl);
console.assert(res.length === 2 && res[2].length === 3 && res[2][1] === 2);
await delay(200);
if (verbose) console.log(this.name, "p_test_list complete")
console.log(this.name, "p_test_list complete")
} catch(err) {
console.log("Exception thrown in ", this.name, "p_test_list:", err.message);
throw err;
}
}
async p_test_kvt(urlexpectedsubstring, verbose=false) {
async p_test_kvt(urlexpectedsubstring) {
/*
Test the KeyValue functionality of any transport that supports it.
urlexpectedsubstring: Some string expected in the publicurl of the table.
*/
if (verbose) {console.log(this.name,"p_test_kvt")}
{console.log(this.name,"p_test_kvt")}
try {
let table = await this.p_newtable("NACL VERIFY:1234567KVT","mytable", {verbose});
let table = await this.p_newtable("NACL VERIFY:1234567KVT","mytable");
let mapurl = table.publicurl;
if (verbose) console.log("newtable=",mapurl);
console.log("newtable=",mapurl);
console.assert(mapurl.includes(urlexpectedsubstring));
await this.p_set(mapurl, "testkey", "testvalue", {verbose});
let res = await this.p_get(mapurl, "testkey", {verbose});
await this.p_set(mapurl, "testkey", "testvalue");
let res = await this.p_get(mapurl, "testkey");
console.assert(res === "testvalue");
await this.p_set(mapurl, "testkey2", {foo: "bar"}, {verbose}); // Try setting to an object
res = await this.p_get(mapurl, "testkey2", {verbose});
await this.p_set(mapurl, "testkey2", {foo: "bar"}); // Try setting to an object
res = await this.p_get(mapurl, "testkey2");
console.assert(res.foo === "bar");
await this.p_set(mapurl, "testkey3", [1,2,3], {verbose}); // Try setting to an array
res = await this.p_get(mapurl, "testkey3", {verbose});
await this.p_set(mapurl, "testkey3", [1,2,3]); // Try setting to an array
res = await this.p_get(mapurl, "testkey3");
console.assert(res[1] === 2);
res = await this.p_keys(mapurl, {verbose});
res = await this.p_keys(mapurl);
console.assert(res.includes("testkey") && res.includes("testkey3") && res.length === 3);
await this.p_delete(mapurl, ["testkey"], {verbose});
res = await this.p_getall(mapurl, {verbose});
if (verbose) console.log("getall=>",res);
await this.p_delete(mapurl, ["testkey"]);
res = await this.p_getall(mapurl);
console.log("getall=>",res);
console.assert(res.testkey2.foo === "bar" && res.testkey3["1"] === 2 && !res.testkey);
await delay(200);
if (verbose) console.log(this.name, "p_test_kvt complete")
console.log(this.name, "p_test_kvt complete")
} catch(err) {
console.log("Exception thrown in ", this.name, "p_test_kvt:", err.message);
throw err;

105
TransportGUN.js
View File

@ -5,6 +5,7 @@ const Url = require('url');
process.env.GUN_ENV = "false";
const Gun = require('gun/gun.js'); // TODO-GUN switchback to gun/gun at some point to get minimized version
require('gun/lib/path.js');
const debuggun = require('debug')('dweb-transports:gun');
// Other Dweb modules
const errors = require('./Errors'); // Standard Dweb Errors
@ -46,8 +47,8 @@ class TransportGUN extends Transport {
gun: object returned when starting GUN
*/
constructor(options, verbose) {
super(options, verbose);
constructor(options) {
super(options);
this.options = options; // Dictionary of options
this.gun = undefined;
this.name = "GUN"; // For console log etc
@ -58,7 +59,7 @@ class TransportGUN extends Transport {
this.status = Transport.STATUS_LOADED;
}
connection(url, verbose) {
connection(url) {
/*
TODO-GUN need to determine what a "rooted" Url is in gun, is it specific to a superpeer for example
Utility function to get Gun object for this URL (note this isn't async)
@ -69,24 +70,24 @@ class TransportGUN extends Transport {
let patharray = url.pathname.split('/'); //[ 'gun', database, table ] but could be arbitrary length path
patharray.shift(); // Loose leading ""
patharray.shift(); // Loose "gun"
if (verbose) console.log("Path=", patharray);
debuggun("path=", patharray);
return this.gun.path(patharray); // Not sure how this could become undefined as it will return g before the path is walked, but if do a lookup on this "g" then should get undefined
}
static setup0(options, verbose) {
static setup0(options) {
/*
First part of setup, create obj, add to Transports but dont attempt to connect, typically called instead of p_setup if want to parallelize connections.
options: { gun: { }, } Set of options - "gun" is used for those to pass direct to Gun
*/
let combinedoptions = Transport.mergeoptions(defaultoptions, options.gun);
console.log("GUN options %o", combinedoptions); // Log even if !verbose
let t = new TransportGUN(combinedoptions, verbose); // Note doesnt start IPFS or OrbitDB
debuggun("options %o", combinedoptions);
let t = new TransportGUN(combinedoptions); // Note doesnt start IPFS or OrbitDB
t.gun = new Gun(t.options); // This doesnt connect, just creates db structure
Transports.addtransport(t);
return t;
}
async p_setup1(verbose, cb) {
async p_setup1(cb) {
/*
This sets up for GUN.
Throws: TODO-GUN-DOC document possible error behavior
@ -95,7 +96,7 @@ class TransportGUN extends Transport {
this.status = Transport.STATUS_STARTING; // Should display, but probably not refreshed in most case
if (cb) cb(this);
//TODO-GUN-TEST - try connect and retrieve info then look at ._.opt.peers
await this.p_status(verbose);
await this.p_status();
} catch(err) {
console.error(this.name,"failed to start",err);
this.status = Transport.STATUS_FAILED;
@ -104,7 +105,7 @@ class TransportGUN extends Transport {
return this;
}
async p_status(verbose) {
async p_status() {
/*
Return an integer for the status of a transport see Transport
*/
@ -114,9 +115,9 @@ class TransportGUN extends Transport {
}
// ===== DATA ======
async p_rawfetch(url, {verbose=false}={}) {
async p_rawfetch(url) {
url = Url.parse(url); // Accept url as string or object
let g = this.connection(url, verbose); // Goes all the way to the key
let g = this.connection(url); // Goes all the way to the key
let val = await this._p_once(g);
if (!val) throw new errors.TransportError("GUN unable to retrieve: "+url.href); // WORKAROUND-GUN-ERRORS - gun doesnt throw errors when it cant find something
let o = typeof val === "string" ? JSON.parse(val) : val; // This looks like it is sync (see same code on p_get and p_rawfetch)
@ -131,7 +132,7 @@ class TransportGUN extends Transport {
// ===== LISTS ========
// noinspection JSCheckFunctionSignatures
async p_rawlist(url, {verbose=false}={}) {
async p_rawlist(url) {
/*
Fetch all the objects in a list, these are identified by the url of the public key used for signing.
(Note this is the 'signedby' parameter of the p_rawadd call, not the 'url' parameter
@ -140,23 +141,22 @@ class TransportGUN extends Transport {
List items may have other data (e.g. reference ids of underlying transport)
:param string url: String with the url that identifies the list.
:param boolean verbose: true for debugging output
:resolve array: An array of objects as stored on the list.
*/
try {
let g = this.connection(url, verbose);
let g = this.connection(url);
let data = await this._p_once(g);
let res = data ? Object.keys(data).filter(k => k !== '_').sort().map(k => data[k]) : []; //See WORKAROUND-GUN-UNDERSCORE
// .filter((obj) => (obj.signedby.includes(url))); // upper layers verify, which filters
if (verbose) console.log("GUN.p_rawlist found", ...utils.consolearr(res));
debuggun("p_rawlist found", ...utils.consolearr(res));
return res;
} catch(err) {
console.log("TransportGUN.p_rawlist failed",err.message);
// Will be logged by Transports
throw(err);
}
}
listmonitor(url, callback, {verbose=false, current=false}={}) {
listmonitor(url, callback, {current=false}={}) {
/*
Setup a callback called whenever an item is added to a list, typically it would be called immediately after a p_rawlist to get any more items not returned by p_rawlist.
@ -164,9 +164,8 @@ class TransportGUN extends Transport {
callback: function(obj) Callback for each new item added to the list
obj is same format as p_rawlist or p_rawreverse
current true if should send list of existing elements
verbose: true for debugging output
*/
let g = this.connection(url, verbose);
let g = this.connection(url);
if (!current) { // See WORKAROUND-GUN-CURRENT have to keep an extra copy to compare for which calls are new.
g.once(data => {
this.monitored = data ? Object.keys(data) : []; // Keep a copy - could actually just keep high water mark unless getting partial knowledge of state of array.
@ -183,7 +182,7 @@ class TransportGUN extends Transport {
}
// noinspection JSCheckFunctionSignatures
async p_rawadd(url, sig, {verbose=false}={}) {
async p_rawadd(url, sig) {
/*
Store a new list item, it should be stored so that it can be retrieved either by "signedby" (using p_rawlist) or
by "url" (with p_rawreverse). The underlying transport does not need to guarantee the signature,
@ -195,33 +194,32 @@ class TransportGUN extends Transport {
urls - array of urls for the object being signed
signature - verifiable signature of date+urls
signedby - urls of public key used for the signature
:param boolean verbose: true for debugging output
:resolve undefined:
*/
// noinspection JSUnresolvedVariable
// Logged by Transports
console.assert(url && sig.urls.length && sig.signature && sig.signedby.length, "TransportGUN.p_rawadd args", url, sig);
if (verbose) console.log("TransportGUN.p_rawadd", typeof url === "string" ? url : url.href, sig);
this.connection(url, verbose)
this.connection(url)
.set( JSON.stringify( sig.preflight( Object.assign({}, sig))));
}
// noinspection JSCheckFunctionSignatures
async p_newlisturls(cl, {verbose=false}={}) {
let u = await this._p_newgun(cl, {verbose});
async p_newlisturls(cl) {
let u = await this._p_newgun(cl);
return [ u, u];
}
//=======KEY VALUE TABLES ========
// noinspection JSMethodCanBeStatic
async _p_newgun(pubkey, {verbose=false}={}) {
async _p_newgun(pubkey) {
if (pubkey.hasOwnProperty("keypair"))
pubkey = pubkey.keypair.signingexport();
// By this point pubkey should be an export of a public key of form xyz:abc where xyz
// specifies the type of public key (NACL VERIFY being the only kind we expect currently)
return `gun:/gun/${encodeURIComponent(pubkey)}`;
}
async p_newdatabase(pubkey, {verbose=false}={}) {
async p_newdatabase(pubkey) {
/*
Request a new database
For GUN it doesnt actually create anything, just generates the URLs
@ -230,11 +228,11 @@ class TransportGUN extends Transport {
returns: {publicurl: "gun:/gun/<publickey>", privateurl: "gun:/gun/<publickey>">
*/
let u = await this._p_newgun(pubkey, {verbose});
let u = await this._p_newgun(pubkey);
return {publicurl: u, privateurl: u};
}
async p_newtable(pubkey, table, {verbose=false}={}) {
async p_newtable(pubkey, table) {
/*
Request a new table
For GUN it doesnt actually create anything, just generates the URLs
@ -242,18 +240,18 @@ class TransportGUN extends Transport {
returns: {publicurl: "gun:/gun/<publickey>/<table>", privateurl: "gun:/gun/<publickey>/<table>">
*/
if (!pubkey) throw new errors.CodingError("p_newtable currently requires a pubkey");
let database = await this.p_newdatabase(pubkey, {verbose});
let database = await this.p_newdatabase(pubkey);
// If have use cases without a database, then call p_newdatabase first
return { privateurl: `${database.privateurl}/${table}`, publicurl: `${database.publicurl}/${table}`} // No action required to create it
}
async p_set(url, keyvalues, value, {verbose=false}={}) { // url = yjs:/yjs/database/table
async p_set(url, keyvalues, value) { // url = yjs:/yjs/database/table
/*
Set key values
keyvalues: string (key) in which case value should be set there OR
object in which case value is ignored
*/
let table = this.connection(url, verbose);
let table = this.connection(url);
if (typeof keyvalues === "string") {
table.path(keyvalues).put(JSON.stringify(value));
} else {
@ -267,8 +265,8 @@ class TransportGUN extends Transport {
}
}
async p_get(url, keys, {verbose=false}={}) {
let table = this.connection(url, verbose);
async p_get(url, keys) {
let table = this.connection(url);
if (Array.isArray(keys)) {
throw new errors.ToBeImplementedError("p_get(url, [keys]) isn't supported - because of ambiguity better to explicitly loop on set of keys or use getall and filter");
/*
@ -284,8 +282,8 @@ class TransportGUN extends Transport {
}
}
async p_delete(url, keys, {verbose=false}={}) {
let table = this.connection(url, verbose);
async p_delete(url, keys) {
let table = this.connection(url);
if (typeof keys === "string") {
table.path(keys).put(null);
} else {
@ -296,34 +294,32 @@ class TransportGUN extends Transport {
//WORKAROUND-GUN-PROMISE suggest p_once as a good single addition
//TODO-GUN expand this to workaround Gun weirdness with errors.
_p_once(gun) { // Note in some cases (e.g. p_getall) this will resolve to a object, others a string/number (p_get)
return new Promise((resolve, reject) => gun.once(resolve));
return new Promise((resolve) => gun.once(resolve));
}
async p_keys(url, {verbose=false}={}) {
let res = await this._p_once(this.connection(url, verbose));
async p_keys(url) {
let res = await this._p_once(this.connection(url));
return Object.keys(res)
.filter(k=> (k !== '_') && (res[k] !== null)); //See WORKAROUND-GUN-UNDERSCORE and WORKAROUND-GUN-DELETE
}
async p_getall(url, {verbose=false}={}) {
let res = await this._p_once(this.connection(url, verbose));
async p_getall(url) {
let res = await this._p_once(this.connection(url));
return Object.keys(res)
.filter(k=> (k !== '_') && res[k] !== null) //See WORKAROUND-GUN-UNDERSCORE and WORKAROUND-GUN-DELETE
.reduce( function(previous, key) { previous[key] = JSON.parse(res[key]); return previous; }, {});
}
async monitor(url, callback, {verbose=false, current=false}={}) {
async monitor(url, callback, {current=false}={}) {
/*
Setup a callback called whenever an item is added to a list, typically it would be called immediately after a p_getall to get any more items not returned by p_getall.
Stack: KVT()|KVT.p_new => KVT.monitor => (a: Transports.monitor => GUN.monitor)(b: dispatchEvent)
url: string Identifier of list (as used by p_rawlist and "signedby" parameter of p_rawadd
callback: function({type, key, value}) Callback for each new item added to the list (type = "set"|"delete")
verbose: boolean - true for debugging output
current Send existing items to the callback as well
*/
let g = this.connection(url, verbose);
let g = this.connection(url);
if (!current) { // See WORKAROUND-GUN-CURRENT have to keep an extra copy to compare for which calls are new.
g.once(data => {
this.monitored = Object.assign({},data); // Make a copy of data (this.monitored = data won't work as just points at same structure)
@ -339,21 +335,22 @@ class TransportGUN extends Transport {
}
}
static async p_test(verbose) {
if (verbose) {console.log("TransportGUN.test")}
static async p_test() {
debuggun("p_test");
try {
let t = this.setup0({}, verbose); //TODO-GUN when works with peers commented out, try passing peers: []
await t.p_setup1(verbose); // Not passing cb yet
await t.p_setup2(verbose); // Not passing cb yet - this one does nothing on GUN
let t = this.setup0({}); //TODO-GUN when works with peers commented out, try passing peers: []
await t.p_setup1(); // Not passing cb yet
await t.p_setup2(); // Not passing cb yet - this one does nothing on GUN
// noinspection JSIgnoredPromiseFromCall
t.p_test_kvt("gun:/gun/NACL", {verbose});
//t.p_test_list("gun:/gun/NACL", {verbose}); //TODO test_list needs fixing to not create a dependency on Signature
t.p_test_kvt("gun:/gun/NACL");
//t.p_test_list("gun:/gun/NACL"); //TODO test_list needs fixing to not create a dependency on Signature
} catch(err) {
console.log("Exception thrown in TransportGUN.test:", err.message);
console.warn("Exception thrown in TransportGUN.test:", err.message);
throw err;
}
}
// noinspection JSUnusedGlobalSymbols
static async demo_bugs() {
let gun = new Gun();
gun.get('foo').get('bar').put('baz');

128
TransportHTTP.js
View File

@ -3,6 +3,7 @@ const Transports = require('./Transports'); // Manage all Transports that are lo
const httptools = require('./httptools'); // Expose some of the httptools so that IPFS can use it as a backup
const Url = require('url');
const stream = require('readable-stream');
const debughttp = require('debug')('dweb-transports:http');
defaulthttpoptions = {
@ -24,8 +25,8 @@ servercommands = { // What the server wants to see to return each of these
class TransportHTTP extends Transport {
constructor(options, verbose) {
super(options, verbose);
constructor(options) {
super(options);
this.options = options;
this.urlbase = options.http.urlbase;
this.supportURLs = ['contenthash', 'http','https'];
@ -35,37 +36,37 @@ class TransportHTTP extends Transport {
this.status = Transport.STATUS_LOADED;
}
static setup0(options, verbose) {
static setup0(options) {
let combinedoptions = Transport.mergeoptions({ http: defaulthttpoptions },options);
try {
let t = new TransportHTTP(combinedoptions, verbose);
let t = new TransportHTTP(combinedoptions);
Transports.addtransport(t);
return t;
} catch (err) {
console.log("Exception thrown in TransportHTTP.p_setup", err.message);
console.error("HTTP unable to setup0", err.message);
throw err;
}
}
async p_setup1(verbose, cb) {
async p_setup1(cb) {
this.status = Transport.STATUS_STARTING;
if (cb) cb(this);
await this.p_status(verbose);
await this.p_status();
if (cb) cb(this);
return this;
}
async p_status(verbose) {
async p_status() {
/*
Return a numeric code for the status of a transport.
*/
try {
this.info = await this.p_info(verbose);
this.info = await this.p_info();
this.status = Transport.STATUS_CONNECTED;
} catch(err) {
console.log(this.name, ": Error in p_status.info",err.message);
console.error(this.name, ": Error in p_status.info",err.message);
this.status = Transport.STATUS_FAILED;
}
return super.p_status(verbose);
return super.p_status();
}
_cmdurl(command) {
@ -84,13 +85,13 @@ class TransportHTTP extends Transport {
Fetch from underlying transport,
Fetch is used both for contenthash requests and table as when passed to SmartDict.p_fetch may not know what we have
url: Of resource - which is turned into the HTTP url in p_httpfetch
opts: {start, end, verbose} see p_GET for documentation
opts: {start, end} see p_GET for documentation
throws: TransportError if fails
*/
//if (!(url && url.includes(':') ))
// throw new errors.CodingError("TransportHTTP.p_rawfetch bad url: "+url);
if (url.href.includes('contenthash//'))
console.error("XXX@91",url)
//if (url.href.includes('contenthash//'))
// console.error("XXX@91", url)
if (((typeof url === "string") ? url : url.href).includes('/getall/table')) {
throw new Error("Probably dont want to be calling p_rawfetch on a KeyValueTable, especially since dont know if its keyvaluetable or subclass"); //TODO-NAMING
return { // I'm not sure what this return would have done - looks half finished to me?
@ -101,39 +102,38 @@ class TransportHTTP extends Transport {
}
}
p_rawlist(url, {verbose=false}={}) {
p_rawlist(url) {
// obj being loaded
// Locate and return a block, based on its url
if (!url) throw new errors.CodingError("TransportHTTP.p_rawlist: requires url");
return httptools.p_GET(this._url(url, servercommands.rawlist), {verbose});
return httptools.p_GET(this._url(url, servercommands.rawlist));
}
rawreverse() { throw new errors.ToBeImplementedError("Undefined function TransportHTTP.rawreverse"); }
async p_rawstore(data, {verbose=false}={}) {
async p_rawstore(data) {
/*
Store data on http server,
data: string
resolves to: {string}: url
throws: TransportError on failure in p_POST > p_httpfetch
*/
//PY: res = self._sendGetPost(True, "rawstore", headers={"Content-Type": "application/octet-stream"}, urlargs=[], data=data, verbose=verbose)
//PY: res = self._sendGetPost(True, "rawstore", headers={"Content-Type": "application/octet-stream"}, urlargs=[], data=data)
console.assert(data, "TransportHttp.p_rawstore: requires data");
let res = await httptools.p_POST(this._cmdurl(servercommands.rawstore), "application/octet-stream", data, verbose); // resolves to URL
let res = await httptools.p_POST(this._cmdurl(servercommands.rawstore), "application/octet-stream", data); // resolves to URL
let parsedurl = Url.parse(res);
let pathparts = parsedurl.pathname.split('/');
return `contenthash:/contenthash/${pathparts.slice(-1)}`
}
p_rawadd(url, sig, {verbose=false}={}) {
//verbose=true;
if (!url || !sig) throw new errors.CodingError("TransportHTTP.p_rawadd: invalid parms",url, sig);
if (verbose) console.log("rawadd", url, sig);
p_rawadd(url, sig) {
// Logged by Transports
if (!url || !sig) throw new errors.CodingError("TransportHTTP.p_rawadd: invalid parms", url, sig);
let value = JSON.stringify(sig.preflight(Object.assign({},sig)))+"\n";
return httptools.p_POST(this._url(url, servercommands.rawadd), "application/json", value, verbose); // Returns immediately
return httptools.p_POST(this._url(url, servercommands.rawadd), "application/json", value); // Returns immediately
}
p_newlisturls(cl, {verbose=false}={}) {
p_newlisturls(cl) {
let u = cl._publicurls.map(urlstr => Url.parse(urlstr))
.find(parsedurl =>
((parsedurl.protocol === "https:" && ["gateway.dweb.me", "dweb.me"].includes(parsedurl.host)
@ -151,7 +151,7 @@ class TransportHTTP extends Transport {
Code disabled until have a chance to test it with <VIDEO> tag etc, problem is that it returns p_createReadStream whch is async
if need sync, look at WebTorrent and how it buffers through a stream which can be returned immediately
*/
async p_f_createReadStream(url, {verbose=false, wanturl=false}={}) {
async p_f_createReadStream(url, {wanturl=false}={}) {
/*
Fetch bytes progressively, using a node.js readable stream, based on a url of the form:
No assumption is made about the data in terms of size or structure.
@ -163,36 +163,36 @@ class TransportHTTP extends Transport {
Node.js readable stream docs: https://nodejs.org/api/stream.html#stream_readable_streams
:param string url: URL of object being retrieved of form magnet:xyzabc/path/to/file (Where xyzabc is the typical magnet uri contents)
:param boolean verbose: true for debugging output
:resolves to: f({start, end}) => stream (The readable stream.)
:throws: TransportError if url invalid - note this happens immediately, not as a catch in the promise
*/
if (verbose) console.log(this.name, "p_f_createreadstream", Url.parse(url).href);
//Logged by Transports
//debughttp("p_f_createreadstream %s", Url.parse(url).href);
try {
let self = this;
if (wanturl) {
return url;
} else {
return function (opts) { return self.p_createReadStream(url, opts, verbose); };
return function (opts) { return self.p_createReadStream(url, opts); };
}
} catch(err) {
console.warn(`p_f_createReadStream failed on ${Url.parse(url).href} ${err.message}`);
//Logged by Transports
//console.warn(`p_f_createReadStream failed on ${Url.parse(url).href} ${err.message}`);
throw(err);
}
}
async createReadStream(url, opts, verbose) {
async createReadStream(url, opts) {
/*
The function, encapsulated and inside another function by p_f_createReadStream (see docs)
NOTE THIS DOESNT WONT WORK FOR <VIDEO> tags, but shouldnt be using it there anyway - reports stream.on an filestream.pipe aren't functions
:param file: Webtorrent "file" as returned by webtorrentfindfile
:param opts: { start: byte to start from; end: optional end byte }
:param boolean verbose: true for debugging output
:resolves to stream: The readable stream.
*/
if (verbose) console.log(this.name, "createreadstream", Url.parse(url).href, opts);
debughttp("createreadstream %s %o", Url.parse(url).href, opts);
let through;
try {
through = new stream.PassThrough();
@ -200,7 +200,7 @@ class TransportHTTP extends Transport {
p_filestream.then(s => s.pipe(through));
return through; // Returns through synchronously, before the pipe is setup
} catch(err) {
console.log(this.name,"createReadStream caught error", err);
console.warn(this.name,"createReadStream caught error", err.message);
if (typeof through.destroy === 'function')
through.destroy(err);
else through.emit('error', err)
@ -210,17 +210,16 @@ class TransportHTTP extends Transport {
}
async p_createReadStream(url, opts, verbose) {
async p_createReadStream(url, opts) {
/*
The function, encapsulated and inside another function by p_f_createReadStream (see docs)
NOTE THIS PROBABLY WONT WORK FOR <VIDEO> tags, but shouldnt be using it there anyway
:param file: Webtorrent "file" as returned by webtorrentfindfile
:param opts: { start: byte to start from; end: optional end byte }
:param boolean verbose: true for debugging output
:resolves to stream: The readable stream.
*/
if (verbose) console.log(this.name, "createreadstream", Url.parse(url).href, opts);
debughttp("createreadstream %s %o", Url.parse(url).href, opts);
try {
return await httptools.p_GET(this._url(url, servercommands.rawfetch), Object.assign({wantstream: true}, opts));
} catch(err) {
@ -235,7 +234,7 @@ class TransportHTTP extends Transport {
// Support for Key-Value pairs as per
// https://docs.google.com/document/d/1yfmLRqKPxKwB939wIy9sSaa7GKOzM5PrCZ4W1jRGW6M/edit#
async p_newdatabase(pubkey, {verbose=false}={}) {
async p_newdatabase(pubkey) {
//if (pubkey instanceof Dweb.PublicPrivate)
if (pubkey.hasOwnProperty("keypair"))
pubkey = pubkey.keypair.signingexport()
@ -246,71 +245,72 @@ class TransportHTTP extends Transport {
}
async p_newtable(pubkey, table, {verbose=false}={}) {
async p_newtable(pubkey, table) {
if (!pubkey) throw new errors.CodingError("p_newtable currently requires a pubkey");
let database = await this.p_newdatabase(pubkey, {verbose});
let database = await this.p_newdatabase(pubkey);
// If have use cases without a database, then call p_newdatabase first
return { privateurl: `${database.privateurl}/${table}`, publicurl: `${database.publicurl}/${table}`} // No action required to create it
}
//TODO-KEYVALUE needs signing with private key of list
async p_set(url, keyvalues, value, {verbose=false}={}) { // url = yjs:/yjs/database/table/key
if (!url || !keyvalues) throw new errors.CodingError("TransportHTTP.p_set: invalid parms",url, keyvalyes);
if (verbose) console.log("p_set", url, keyvalues, value);
async p_set(url, keyvalues, value) { // url = yjs:/yjs/database/table/key
if (!url || !keyvalues) throw new errors.CodingError("TransportHTTP.p_set: invalid parms", url, keyvalyes);
// Logged by Transports
//debughttp("p_set %o %o %o", url, keyvalues, value);
if (typeof keyvalues === "string") {
let kv = JSON.stringify([{key: keyvalues, value: value}]);
await httptools.p_POST(this._url(url, servercommands.set), "application/json", kv, verbose); // Returns immediately
await httptools.p_POST(this._url(url, servercommands.set), "application/json", kv); // Returns immediately
} else {
let kv = JSON.stringify(Object.keys(keyvalues).map((k) => ({"key": k, "value": keyvalues[k]})));
await httptools.p_POST(this._url(url, servercommands.set), "application/json", kv, verbose); // Returns immediately
await httptools.p_POST(this._url(url, servercommands.set), "application/json", kv); // Returns immediately
}
}
_keyparm(key) {
return `key=${encodeURIComponent(key)}`
}
async p_get(url, keys, {verbose=false}={}) {
async p_get(url, keys) {
if (!url && keys) throw new errors.CodingError("TransportHTTP.p_get: requires url and at least one key");
let parmstr =Array.isArray(keys) ? keys.map(k => this._keyparm(k)).join('&') : this._keyparm(keys)
let res = await httptools.p_GET(this._url(url, servercommands.get, parmstr), {verbose});
let res = await httptools.p_GET(this._url(url, servercommands.get, parmstr));
return Array.isArray(keys) ? res : res[keys]
}
async p_delete(url, keys, {verbose=false}={}) {
async p_delete(url, keys) {
if (!url && keys) throw new errors.CodingError("TransportHTTP.p_get: requires url and at least one key");
let parmstr = keys.map(k => this._keyparm(k)).join('&');
await httptools.p_GET(this._url(url, servercommands.delete, parmstr), {verbose});
await httptools.p_GET(this._url(url, servercommands.delete, parmstr));
}
async p_keys(url, {verbose=false}={}) {
async p_keys(url) {
if (!url && keys) throw new errors.CodingError("TransportHTTP.p_get: requires url and at least one key");
return await httptools.p_GET(this._url(url, servercommands.keys), {verbose});
return await httptools.p_GET(this._url(url, servercommands.keys));
}
async p_getall(url, {verbose=false}={}) {
async p_getall(url) {
if (!url && keys) throw new errors.CodingError("TransportHTTP.p_get: requires url and at least one key");
return await httptools.p_GET(this._url(url, servercommands.getall), {verbose});
return await httptools.p_GET(this._url(url, servercommands.getall));
}
/* Make sure doesnt shadow regular p_rawfetch
async p_rawfetch(url, verbose) {
async p_rawfetch(url) {
return {
table: "keyvaluetable",
_map: await this.p_getall(url, {verbose})
_map: await this.p_getall(url)
}; // Data struc is ok as SmartDict.p_fetch will pass to KVT constructor
}
*/
p_info(verbose) { return httptools.p_GET(`${this.urlbase}/info`, {verbose}); }
p_info() { return httptools.p_GET(`${this.urlbase}/info`); }
static async p_test(opts={}, verbose=false) {
if (verbose) {console.log("TransportHTTP.test")}
static async p_test(opts={}) {
{console.log("TransportHTTP.test")}
try {
let transport = await this.p_setup(opts, verbose);
if (verbose) console.log("HTTP connected");
let res = await transport.p_info(verbose);
if (verbose) console.log("TransportHTTP info=",res);
res = await transport.p_status(verbose);
let transport = await this.p_setup(opts);
console.log("HTTP connected");
let res = await transport.p_info();
console.log("TransportHTTP info=",res);
res = await transport.p_status();
console.assert(res === Transport.STATUS_CONNECTED);
await transport.p_test_kvt("NACL%20VERIFY", verbose);
await transport.p_test_kvt("NACL%20VERIFY");
} catch(err) {
console.log("Exception thrown in TransportHTTP.test:", err.message);
throw err;

105
TransportIPFS.js
View File

@ -4,6 +4,7 @@ See https://github.com/ipfs/js-ipfs but note its often out of date relative to t
*/
const httptools = require('./httptools'); // Expose some of the httptools so that IPFS can use it as a backup
const debugipfs = require('debug')('dweb-transports:ipfs');
// IPFS components
const IPFS = require('ipfs');
@ -54,8 +55,8 @@ class TransportIPFS extends Transport {
TODO - this is not complete
*/
constructor(options, verbose) {
super(options, verbose);
constructor(options) {
super(options);
this.ipfs = undefined; // Undefined till start IPFS
this.options = options; // Dictionary of options
this.name = "IPFS"; // For console log etc
@ -76,7 +77,7 @@ class TransportIPFS extends Transport {
}}}
}
*/
p_ipfsstart(verbose) {
p_ipfsstart() {
/*
Just start IPFS - not Y (note used with "yarrays" and will be used for non-IPFS list management)
Note - can't figure out how to use async with this, as we resolve the promise based on the event callback
@ -91,45 +92,47 @@ class TransportIPFS extends Transport {
this.ipfs.on('error', (err) => reject(err));
})
.then(() => self.ipfs.version())
.then((version) => console.log('IPFS READY',version))
.then((version) => debugipfs('ready %o',version))
.catch((err) => {
console.log("Error caught in p_ipfsstart");
console.warn("IPFS p_ipfsstart failed", err.message);
throw(err);
});
}
static setup0(options, verbose) {
static setup0(options) {
/*
First part of setup, create obj, add to Transports but dont attempt to connect, typically called instead of p_setup if want to parallelize connections.
*/
const combinedoptions = Transport.mergeoptions(defaultoptions, options.ipfs);
if (verbose) console.log("IPFS loading options %o", combinedoptions);
const t = new TransportIPFS(combinedoptions, verbose); // Note doesnt start IPFS
debugipfs("setup options=%o", combinedoptions);
const t = new TransportIPFS(combinedoptions); // Note doesnt start IPFS
Transports.addtransport(t);
return t;
}
async p_setup1(verbose, cb) {
async p_setup1(cb) {
try {
if (verbose) console.log("IPFS starting and connecting");
// Logged by Transports
this.status = Transport.STATUS_STARTING; // Should display, but probably not refreshed in most case
if (cb) cb(this);
await this.p_ipfsstart(verbose); // Throws Error("websocket error") and possibly others.
this.status = await this.p_status(verbose);
await this.p_ipfsstart(); // Throws Error("websocket error") and possibly others.
this.status = await this.p_status();
} catch(err) {
// Logged by Transports
console.error(this.name, "failed to connect", err);
this.status = Transport.STATUS_FAILED;
// Dont throw an error, allow other transports to complete setup
}
if (cb) cb(this);
return this;
}
async p_status(verbose) {
async p_status() {
/*
Return a numeric code for the status of a transport.
*/
this.status = (await this.ipfs.isOnline()) ? Transport.STATUS_CONNECTED : Transport.STATUS_FAILED;
return super.p_status(verbose);
return super.p_status();
}
// Everything else - unless documented here - should be opaque to the actual structure of a CID
@ -164,7 +167,7 @@ class TransportIPFS extends Transport {
throw new errors.TransportError("TransportIPFS.cidFrom not supporting paths in url yet, should be dweb: or ipfs:/ipfs/...: " + url.href);
return new CID(patharr[2]);
} else {
throw new errors.CodingError("TransportIPFS.cidFrom: Cant convert url",url);
throw new errors.CodingError("TransportIPFS.cidFrom: Cant convert url", url);
}
}
@ -214,7 +217,7 @@ class TransportIPFS extends Transport {
throw new errors.CodingError(`Cant turn ${url} into a multihash`);
}
async p_rawfetch(url, {verbose=false, timeoutMS=60000, relay=false}={}) {
async p_rawfetch(url, {timeoutMS=60000, relay=false}={}) {
/*
Fetch some bytes based on a url of the form ipfs:/ipfs/Qm..... or ipfs:/ipfs/z.... .
No assumption is made about the data in terms of size or structure, nor can we know whether it was created with dag.put or ipfs add or http /api/v0/add/
@ -224,11 +227,10 @@ class TransportIPFS extends Transport {
There may also be need for a streaming version of this call, at this point undefined since we havent (currently) got a use case..
:param string url: URL of object being retrieved {ipfs|dweb}:/ipfs/<cid> or /
:param boolean verbose: true for debugging output
:resolve buffer: Return the object being fetched. (may in the future return a stream and buffer externally)
:throws: TransportError if url invalid - note this happens immediately, not as a catch in the promise
*/
if (verbose) console.log("IPFS p_rawfetch", Url.parse(url).href);
// Attempt logged by Transports
if (!url) throw new errors.CodingError("TransportIPFS.p_rawfetch: requires url");
const cid = TransportIPFS.cidFrom(url); // Throws TransportError if url bad
const ipfspath = TransportIPFS.ipfsFrom(url) // Need because dag.get has different requirement than file.cat
@ -241,61 +243,45 @@ class TransportIPFS extends Transport {
throw new errors.TransportError("Not yet supporting paths in p_rawfetch");
} //TODO-PATH
let buff;
//if (res.value instanceof DAGNode) { // Its file or something added with the HTTP API for example, TODO not yet handling multiple files
if (res.value.constructor.name === "DAGNode") { // Kludge to replace above, as its not matching the type against the "require" above.
if (verbose) console.log("IPFS p_rawfetch looks like its a file", Url.parse(url).href);
//console.log("Case a or b" - we can tell the difference by looking at (res.value._links.length > 0) but dont need to
// as since we dont know if we are on node or browser best way is to try the files.cat and if it fails try the block to get an approximate file);
// Works on Node, but fails on Chrome, cant figure out how to get data from the DAGNode otherwise (its the wrong size)
// We retrieved a DAGNode, call files.cat (the node will come from the cache quickly)
buff = await this.ipfs.files.cat(ipfspath); //See js-ipfs v0.27 version and https://github.com/ipfs/js-ipfs/issues/1229 and https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/FILES.md#cat
/* Was needed on v0.26, not on v0.27
if (buff.length === 0) { // Hit the Chrome bug
// This will get a file padded with ~14 bytes - 4 at front, 4 at end and cant find the other 6 !
// but it seems to work for PDFs which is what I'm testing on.
if (verbose) console.log("Kludge alert - files.cat fails in Chrome, trying block.get");
let blk = await this.promisified.ipfs.block.get(cid);
buff = blk.data;
}
END of v0.26 version */
} else { //c: not a file
debugipfs("Found a raw IPFS block (unusual) - not a DAGNode - handling as such");
buff = res.value;
}
if (verbose) console.log(`IPFS fetched ${buff.length} from ${ipfspath}`);
// Success logged by Transports
return buff;
} catch (err) { // TimeoutError or could be some other error from IPFS etc
console.log("Caught misc error in TransportIPFS.p_rawfetch trying IPFS", err.message);
debugipfs("Caught error '%s' fetching via IPFS, trying IPFS HTTP gateway", err.message);
try {
let ipfsurl = TransportIPFS.ipfsGatewayFrom(url);
return await utils.p_timeout(
httptools.p_GET(ipfsurl), // Returns a buffer
timeoutMS, "Timed out IPFS fetch of "+ipfsurl)
} catch (err) {
console.log("Caught misc error in TransportIPFS.p_rawfetch trying gateway", err.message);
// Failure logged by Transports:
//debugipfs("Failed to retrieve from gateway: %s", err.message);
throw err;
}
}
}
async p_rawstore(data, {verbose}) {
async p_rawstore(data) {
/*
Store a blob of data onto the decentralised transport.
Returns a promise that resolves to the url of the data
:param string|Buffer data: Data to store - no assumptions made to size or content
:param boolean verbose: true for debugging output
:resolve string: url of data stored
*/
console.assert(data, "TransportIPFS.p_rawstore: requires data");
const buf = (data instanceof Buffer) ? data : new Buffer(data);
//return this.promisified.ipfs.block.put(buf).then((block) => block.cid)
//https://github.com/ipfs/interface-ipfs-core/blob/master/SPEC/DAG.md#dagput
//let res = await this.ipfs.dag.put(buf,{ format: 'dag-cbor', hashAlg: 'sha2-256' });
const res = (await this.ipfs.files.add(buf,{ "cid-version": 1, hashAlg: 'sha2-256'}))[0];
return TransportIPFS.urlFrom(res);
}
/* OLD WAY Based on https://github.com/ipfs/js-ipfs/pull/1231/files
/* OLD WAY Based on https://github.com/ipfs/js-ipfs/pull/1231/files TODO-IPFS repurpose this to add byte range to fetch
async p_offsetStream(stream, links, startByte, endByte) {
let streamPosition = 0
@ -333,9 +319,8 @@ class TransportIPFS extends Transport {
console.log(err.message);
}
}
async p_f_createReadStream(url, {verbose=false}={}) { // Asynchronously return a function that can be used in createReadStream
verbose = true;
if (verbose) console.log("p_f_createReadStream",url);
async p_f_createReadStream(url) { // Asynchronously return a function that can be used in createReadStream
if () console.log("p_f_createReadStream", url);
const mh = TransportIPFS.multihashFrom(url);
const links = await this.ipfs.object.links(mh);
let throughstream; //Holds pointer to stream between calls.
@ -360,7 +345,7 @@ class TransportIPFS extends Transport {
}
*/
async p_f_createReadStream(url, {verbose=false, wanturl=false}={}) {
async p_f_createReadStream(url, {wanturl=false}={}) {
/*
Fetch bytes progressively, using a node.js readable stream, based on a url of the form:
No assumption is made about the data in terms of size or structure.
@ -374,11 +359,11 @@ class TransportIPFS extends Transport {
:param string url: URL of object being retrieved of form:
magnet:xyzabc/path/to/file (Where xyzabc is the typical magnet uri contents)
ipfs:/ipfs/Q123
:param boolean verbose: true for debugging output
:resolves to: f({start, end}) => stream (The readable stream.)
:throws: TransportError if url invalid - note this happens immediately, not as a catch in the promise
*/
if (verbose) console.log(this.name, "p_f_createreadstream %o", url);
// Logged by Transports;
//debugipfs("p_f_createreadstream %o", url);
let stream;
try {
let multihash = url.pathname.split('/ipfs/')[1]
@ -392,11 +377,10 @@ class TransportIPFS extends Transport {
/*
The function, encapsulated and inside another function by p_f_createReadStream (see docs)
:param opts: { start: byte to start from; end: optional end byte }
:param boolean verbose: true for debugging output
:returns stream: The readable stream.
FOR IPFS this is copied and adapted from git repo js-ipfs/examples/browser-readablestream/index.js
*/
if (verbose) console.log("TransportIPFS createreadstream %o %o", multihash, opts);
debugipfs("createReadStream %o %o", multihash, opts || "" );
const start = opts ? opts.start : 0
// The videostream library does not always pass an end byte but when
@ -410,6 +394,8 @@ class TransportIPFS extends Transport {
}
// This stream will contain the requested bytes
// For debugging used a known good IPFS video
//let fakehash="QmedXJYwvNSJFRMVFuJt7BfCMcJwPoqJgqN3U2MYxHET5a"
//console.log("XXX@IPFS.p_f_createReadStream faking call to",multihash, "with", fakehash)
//multihash=fakehash;
@ -433,32 +419,33 @@ class TransportIPFS extends Transport {
if (stream && stream.destroy) {
stream.destroy()
}
console.log(`p_f_createReadStream failed on ${url} ${err.message}`);
// Error logged by Transports
//console.log(`p_f_createReadStream failed on ${url} ${err.message}`);
throw(err);
};
}
static async p_test(opts, verbose) {
if (verbose) {console.log("TransportIPFS.test")}
static async p_test(opts) {
{console.log("TransportIPFS.test")}
try {
const transport = await this.p_setup(opts, verbose); // Assumes IPFS already setup
if (verbose) console.log(transport.name,"setup");
const res = await transport.p_status(verbose);
const transport = await this.p_setup(opts); // Assumes IPFS already setup
console.log(transport.name,"setup");
const res = await transport.p_status();
console.assert(res === Transport.STATUS_CONNECTED)
let urlqbf;
const qbf = "The quick brown fox";
const qbf_url = "ipfs:/ipfs/zdpuAscRnisRkYnEyJAp1LydQ3po25rCEDPPEDMymYRfN1yPK"; // Expected url
const testurl = "1114"; // Just a predictable number can work with
const url = await transport.p_rawstore(qbf, {verbose});
if (verbose) console.log("rawstore returned", url);
const url = await transport.p_rawstore(qbf);
console.log("rawstore returned", url);
const newcid = TransportIPFS.cidFrom(url); // Its a CID which has a buffer in it
console.assert(url === qbf_url, "url should match url from rawstore");
const cidmultihash = url.split('/')[2]; // Store cid from first block in form of multihash
const newurl = TransportIPFS.urlFrom(newcid);
console.assert(url === newurl, "Should round trip");
urlqbf = url;
const data = await transport.p_rawfetch(urlqbf, {verbose});
const data = await transport.p_rawfetch(urlqbf);
console.assert(data.toString() === qbf, "Should fetch block stored above");
//console.log("TransportIPFS test complete");
return transport

69
TransportWEBTORRENT.js
View File

@ -9,6 +9,7 @@ Y Lists have listeners and generate events - see docs at ...
const WebTorrent = require('webtorrent');
const stream = require('readable-stream');
const Url = require('url');
const debugwt = require('debug')('dweb-transports:webtorrent');
// Other Dweb modules
const errors = require('./Errors'); // Standard Dweb Errors
@ -26,8 +27,8 @@ class TransportWEBTORRENT extends Transport {
webtorrent: object returned when starting webtorrent
*/
constructor(options, verbose) {
super(options, verbose);
constructor(options) {
super(options);
this.webtorrent = undefined; // Undefined till start WebTorrent
this.options = options; // Dictionary of options
this.name = "WEBTORRENT"; // For console log etc
@ -36,7 +37,7 @@ class TransportWEBTORRENT extends Transport {
this.status = Transport.STATUS_LOADED;
}
p_webtorrentstart(verbose) {
p_webtorrentstart() {
/*
Start WebTorrent and wait until for ready.
*/
@ -44,7 +45,7 @@ class TransportWEBTORRENT extends Transport {
return new Promise((resolve, reject) => {
this.webtorrent = new WebTorrent(this.options);
this.webtorrent.once("ready", () => {
console.log("WEBTORRENT READY");
debugwt("ready");
resolve();
});
this.webtorrent.once("error", (err) => reject(err));
@ -54,33 +55,33 @@ class TransportWEBTORRENT extends Transport {
})
}
static setup0(options, verbose) {
static setup0(options) {
/*
First part of setup, create obj, add to Transports but dont attempt to connect, typically called instead of p_setup if want to parallelize connections.
*/
let combinedoptions = Transport.mergeoptions(defaultoptions, options.webtorrent);
if (verbose) console.log("WebTorrent options %o", combinedoptions); // Dont normally log options as its long
let t = new TransportWEBTORRENT(combinedoptions, verbose);
debugwt("setup0: options=%o", combinedoptions);
let t = new TransportWEBTORRENT(combinedoptions);
Transports.addtransport(t);
return t;
}
async p_setup1(verbose, cb) {
async p_setup1(cb) {
try {
this.status = Transport.STATUS_STARTING;
if (cb) cb(this);
await this.p_webtorrentstart(verbose);
await this.p_status(verbose);
await this.p_webtorrentstart();
await this.p_status();
} catch(err) {
console.error(this.name, "failed to connect", err);
console.error(this.name, "failed to connect", err.message);
this.status = Transport.STATUS_FAILED;
}
if (cb) cb(this);
return this;
}
async p_status(verbose) {
async p_status() {
/*
Return a string for the status of a transport. No particular format, but keep it short as it will probably be in a small area of the screen.
*/
@ -91,7 +92,7 @@ class TransportWEBTORRENT extends Transport {
} else {
this.status = Transport.STATUS_FAILED;
}
return super.p_status(verbose);
return super.p_status();
}
webtorrentparseurl(url) {
@ -163,7 +164,7 @@ class TransportWEBTORRENT extends Transport {
return file;
}
p_rawfetch(url, {verbose=false}={}) {
p_rawfetch(url) {
/*
Fetch some bytes based on a url of the form:
@ -174,13 +175,11 @@ class TransportWEBTORRENT extends Transport {
No assumption is made about the data in terms of size or structure. Returns a new Promise that resolves to a buffer.
:param string url: URL of object being retrieved
:param boolean verbose: true for debugging output
:resolve buffer: Return the object being fetched.
:throws: TransportError if url invalid - note this happens immediately, not as a catch in the promise
*/
return new Promise((resolve, reject) => {
if (verbose) console.log("WebTorrent p_rawfetch", url);
// Logged by Transports
const { torrentId, path } = this.webtorrentparseurl(url);
this.p_webtorrentadd(torrentId)
.then((torrent) => {
@ -218,13 +217,13 @@ class TransportWEBTORRENT extends Transport {
}
return file
} catch(err) {
console.log(`p_fileFrom failed on ${url} ${err.message}`);
// Logged by Transports
throw(err);
}
}
async p_f_createReadStream(url, {verbose=false, wanturl=false}={}) {
async p_f_createReadStream(url, {wanturl=false}={}) {
/*
Fetch bytes progressively, using a node.js readable stream, based on a url of the form:
No assumption is made about the data in terms of size or structure.
@ -236,35 +235,33 @@ class TransportWEBTORRENT extends Transport {
Node.js readable stream docs: https://nodejs.org/api/stream.html#stream_readable_streams
:param string url: URL of object being retrieved of form magnet:xyzabc/path/to/file (Where xyzabc is the typical magnet uri contents)
:param boolean verbose: true for debugging output
:resolves to: f({start, end}) => stream (The readable stream.)
:throws: TransportError if url invalid - note this happens immediately, not as a catch in the promise
*/
if (verbose) console.log(this.name, "p_f_createreadstream", Url.parse(url).href);
// Logged by Transports
try {
let filet = await this._p_fileTorrentFromUrl(url);
let self = this;
if (wanturl) {
return url;
} else {
return function (opts) { return self.createReadStream(filet, opts, verbose); };
return function (opts) { return self.createReadStream(filet, opts); };
}
} catch(err) {
console.log(`p_f_createReadStream failed on ${url} ${err.message}`);
// Logged by Transports
throw(err);
}
}
createReadStream(file, opts, verbose) {
createReadStream(file, opts) {
/*
The function, encapsulated and inside another function by p_f_createReadStream (see docs)
:param file: Webtorrent "file" as returned by webtorrentfindfile
:param opts: { start: byte to start from; end: optional end byte }
:param boolean verbose: true for debugging output
:returns stream: The readable stream.
*/
if (verbose) console.log(this.name, "createreadstream", file.name, opts);
debugwt("createReadStream %s %o", file.name, opts);
let through;
try {
through = new stream.PassThrough();
@ -272,20 +269,20 @@ class TransportWEBTORRENT extends Transport {
fileStream.pipe(through);
return through;
} catch(err) {
console.log("TransportWEBTORRENT caught error", err);
debugwt("createReadStream error %s", err);
if (typeof through.destroy === 'function')
through.destroy(err);
else through.emit('error', err)
}
}
async p_createReadableStream(url, opts, verbose) {
async p_createReadableStream(url, opts) {
//Return a readable stream (suitable for a HTTP response) from a node type stream from webtorrent.
// This is used by dweb-serviceworker for WebTorrent only
let filet = await this._p_fileTorrentFromUrl(url);
return new ReadableStream({
start (controller) {
console.log('start', url, opts);
debugwt("start %s %o", url, opts);
// Create a webtorrent file stream
const filestream = filet.createReadStream(opts);
// When data comes out of webtorrent node.js style stream, put it into the WHATWG stream
@ -304,21 +301,21 @@ class TransportWEBTORRENT extends Transport {
static async p_test(opts, verbose) {
static async p_test(opts) {
try {
let transport = await this.p_setup(opts, verbose); // Assumes IPFS already setup
if (verbose) console.log(transport.name, "setup");
let res = await transport.p_status(verbose);
let transport = await this.p_setup(opts); // Assumes IPFS already setup
console.log(transport.name, "p_test setup", opts, "complete");
let res = await transport.p_status();
console.assert(res === Transport.STATUS_CONNECTED);
// Creative commons torrent, copied from https://webtorrent.io/free-torrents
let bigBuckBunny = 'magnet:?xt=urn:btih:dd8255ecdc7ca55fb0bbf81323d87062db1f6d1c&dn=Big+Buck+Bunny&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fbig-buck-bunny.torrent/Big Buck Bunny.en.srt';
let data1 = await transport.p_rawfetch(bigBuckBunny, {verbose});
let data1 = await transport.p_rawfetch(bigBuckBunny);
data1 = data1.toString();
assertData(data1);
const stream = await transport.createReadStream(bigBuckBunny, verbose);
const stream = await transport.createReadStream(bigBuckBunny);
const chunks = [];
stream.on("data", (chunk) => {
@ -339,7 +336,7 @@ class TransportWEBTORRENT extends Transport {
console.assert(data.length, 129, "'Big Buck Bunny.en.srt' was " + data.length);
}
} catch (err) {
console.log("Exception thrown in TransportWEBTORRENT.p_test:", err.message);
console.log("Exception thrown in", transport.name, "p_test():", err.message);
throw err;
}
}

116
TransportYJS.js
View File

@ -4,6 +4,7 @@ This Transport layers builds on the YJS DB and uses IPFS as its transport.
Y Lists have listeners and generate events - see docs at ...
*/
const Url = require('url');
const debugyjs = require('debug')('dweb-transports:yjs');
//const Y = require('yjs/dist/y.js'); // Explicity require of dist/y.js to get around a webpack warning but causes different error in YJS
const Y = require('yjs'); // Explicity require of dist/y.js to get around a webpack warning
@ -41,8 +42,8 @@ class TransportYJS extends Transport {
Fields: TODO document this
*/
constructor(options, verbose) {
super(options, verbose);
constructor(options) {
super(options);
this.options = options; // Dictionary of options
this.name = "YJS"; // For console log etc
this.supportURLs = ['yjs'];
@ -51,7 +52,7 @@ class TransportYJS extends Transport {
this.status = Transport.STATUS_LOADED;
}
async p__y(url, opts, verbose) {
async p__y(url, opts) {
/*
Utility function to get Y for this URL with appropriate options and open a new connection if not already
@ -63,20 +64,20 @@ class TransportYJS extends Transport {
console.assert(url.startsWith("yjs:/yjs/"));
try {
if (this.yarrays[url]) {
if (verbose) console.log("Found Y for", url);
//debugyjs("Found Y for %s", url);
return this.yarrays[url];
} else {
let options = Transport.mergeoptions(this.options, {connector: {room: url}}, opts); // Copies options, ipfs will be set already
if (verbose) console.log("Creating Y for", url); //"options=",options);
//debugyjs("Creating Y for %s", url);
return this.yarrays[url] = await Y(options);
}
} catch(err) {
console.log("Failed to initialize Y");
console.error("Failed to initialize Y", err.message);
throw err;
}
}
async p__yarray(url, verbose) {
async p__yarray(url) {
/*
Utility function to get Yarray for this URL and open a new connection if not already
url: URL string to find list of
@ -84,7 +85,7 @@ class TransportYJS extends Transport {
*/
return this.p__y(url, { share: {array: "Array"}}); // Copies options, ipfs will be set already
}
async p_connection(url, verbose) {
async p_connection(url) {
/*
Utility function to get Yarray for this URL and open a new connection if not already
url: URL string to find list of
@ -95,18 +96,18 @@ class TransportYJS extends Transport {
static setup0(options, verbose) {
static setup0(options) {
/*
First part of setup, create obj, add to Transports but dont attempt to connect, typically called instead of p_setup if want to parallelize connections.
*/
let combinedoptions = Transport.mergeoptions(defaultoptions, options.yjs);
if (verbose) console.log("YJS options %o", combinedoptions); // Log even if !verbose
let t = new TransportYJS(combinedoptions, verbose); // Note doesnt start IPFS or Y
debugyjs("YJS options %o", combinedoptions);
let t = new TransportYJS(combinedoptions); // Note doesnt start IPFS or Y
Transports.addtransport(t);
return t;
}
async p_setup2(verbose, cb) {
async p_setup2(cb) {
/*
This sets up for Y connections, which are opened each time a resource is listed, added to, or listmonitored.
p_setup2 is defined because IPFS will have started during the p_setup1 phase.
@ -115,9 +116,9 @@ class TransportYJS extends Transport {
try {
this.status = Transport.STATUS_STARTING; // Should display, but probably not refreshed in most case
if (cb) cb(this);
this.options.connector.ipfs = Transports.ipfs(verbose).ipfs; // Find an IPFS to use (IPFS's should be starting in p_setup1)
this.options.connector.ipfs = Transports.ipfs().ipfs; // Find an IPFS to use (IPFS's should be starting in p_setup1)
this.yarrays = {};
await this.p_status(verbose);
await this.p_status();
} catch(err) {
console.error(this.name,"failed to start",err);
this.status = Transport.STATUS_FAILED;
@ -126,18 +127,18 @@ class TransportYJS extends Transport {
return this;
}
async p_status(verbose) {
async p_status() {
/*
Return a string for the status of a transport. No particular format, but keep it short as it will probably be in a small area of the screen.
For YJS, its online if IPFS is.
*/
this.status = (await this.options.connector.ipfs.isOnline()) ? Transport.STATUS_CONNECTED : Transport.STATUS_FAILED;
return super.p_status(verbose);
return super.p_status();
}
// ======= LISTS ========
async p_rawlist(url, {verbose=false}={}) {
async p_rawlist(url) {
/*
Fetch all the objects in a list, these are identified by the url of the public key used for signing.
(Note this is the 'signedby' parameter of the p_rawadd call, not the 'url' parameter
@ -146,29 +147,29 @@ class TransportYJS extends Transport {
List items may have other data (e.g. reference ids of underlying transport)
:param string url: String with the url that identifies the list.
:param boolean verbose: true for debugging output
:resolve array: An array of objects as stored on the list.
*/
try {
let y = await this.p__yarray(url, verbose);
let y = await this.p__yarray(url);
let res = y.share.array.toArray();
// .filter((obj) => (obj.signedby.includes(url))); Cant filter since url is the YJS URL, not the URL of the CL that signed it. (upper layers verify, which filters)
if (verbose) console.log("p_rawlist found", ...utils.consolearr(res));
//Logged by Transports
//debugyjs("p_rawlist found %o", res);
return res;
} catch(err) {
console.log("TransportYJS.p_rawlist failed",err.message);
//Logged by Transports
// console.log("TransportYJS.p_rawlist failed",err.message);
throw(err);
}
}
listmonitor(url, callback, {verbose=false, current=false}={}) {
listmonitor(url, callback, {current=false}={}) {
/*
Setup a callback called whenever an item is added to a list, typically it would be called immediately after a p_rawlist to get any more items not returned by p_rawlist.
:param url: string Identifier of list (as used by p_rawlist and "signedby" parameter of p_rawadd
:param callback: function(obj) Callback for each new item added to the list
obj is same format as p_rawlist or p_rawreverse
:param verbose: boolean - true for debugging output
*/
let y = this.yarrays[typeof url === "string" ? url : url.href];
console.assert(y,"Should always exist before calling listmonitor - async call p__yarray(url) to create");
@ -177,7 +178,7 @@ class TransportYJS extends Transport {
}
y.share.array.observe((event) => {
if (event.type === 'insert') { // Currently ignoring deletions.
if (verbose) console.log('resources inserted', event.values);
debugyjs('resources inserted %o', event.values);
//cant filter because url is YJS local, not signer, callback should filter
//event.values.filter((obj) => obj.signedby.includes(url)).map(callback);
event.values.map(callback);
@ -192,13 +193,12 @@ class TransportYJS extends Transport {
Returns a promise that resolves to the list.
:param string url: String with the url that identifies the object put on a list.
:param boolean verbose: true for debugging output
:resolve array: An array of objects as stored on the list.
*/
//TODO-REVERSE this needs implementing once list structure on IPFS more certain
throw new errors.ToBeImplementedError("Undefined function TransportYJS.rawreverse"); }
async p_rawadd(url, sig, {verbose=false}={}) {
async p_rawadd(url, sig) {
/*
Store a new list item, it should be stored so that it can be retrieved either by "signedby" (using p_rawlist) or
by "url" (with p_rawreverse). The underlying transport does not need to guarantee the signature,
@ -210,17 +210,17 @@ class TransportYJS extends Transport {
urls - array of urls for the object being signed
signature - verifiable signature of date+urls
signedby - urls of public key used for the signature
:param boolean verbose: true for debugging output
:resolve undefined:
*/
// Logged by Transports
//debugyjs("TransportYJS.p_rawadd %o %o", url.href, sig);
console.assert(url && sig.urls.length && sig.signature && sig.signedby.length, "TransportYJS.p_rawadd args", url, sig);
if (verbose) console.log("TransportYJS.p_rawadd", typeof url === "string" ? url : url.href, sig);
let value = sig.preflight(Object.assign({}, sig));
let y = await this.p__yarray(url, verbose);
let y = await this.p__yarray(url);
y.share.array.push([value]);
}
p_newlisturls(cl, {verbose=false}={}) {
p_newlisturls(cl) {
let u = cl._publicurls.map(urlstr => Url.parse(urlstr))
.find(parsedurl =>
(parsedurl.protocol === "ipfs" && parsedurl.pathname.includes('/ipfs/'))
@ -233,7 +233,7 @@ class TransportYJS extends Transport {
// ======= KEY VALUE TABLES ========
async p_newdatabase(pubkey, {verbose=false}={}) {
async p_newdatabase(pubkey) {
//if (pubkey instanceof Dweb.PublicPrivate)
if (pubkey.hasOwnProperty("keypair"))
pubkey = pubkey.keypair.signingexport();
@ -245,27 +245,27 @@ class TransportYJS extends Transport {
//TODO maybe change the listmonitor / monitor code for to use "on" and the structure of PP.events
//TODO but note https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Proxy about Proxy which might be suitable, prob not as doesnt map well to lists
async p_newtable(pubkey, table, {verbose=false}={}) {
async p_newtable(pubkey, table) {
if (!pubkey) throw new errors.CodingError("p_newtable currently requires a pubkey");
let database = await this.p_newdatabase(pubkey, {verbose});
let database = await this.p_newdatabase(pubkey);
// If have use cases without a database, then call p_newdatabase first
return { privateurl: `${database.privateurl}/${table}`, publicurl: `${database.publicurl}/${table}`} // No action required to create it
}
async p_set(url, keyvalues, value, {verbose=false}={}) { // url = yjs:/yjs/database/table
async p_set(url, keyvalues, value) { // url = yjs:/yjs/database/table
/*
Set key values
keyvalues: string (key) in which case value should be set there OR
object in which case value is ignored
*/
let y = await this.p_connection(url, verbose);
let y = await this.p_connection(url);
if (typeof keyvalues === "string") {
y.share.map.set(keyvalues, JSON.stringify(value));
} else {
Object.keys(keyvalues).map((key) => y.share.map.set(key, keyvalues[key]));
}
}
_p_get(y, keys, {verbose=false}={}) {
_p_get(y, keys) {
if (Array.isArray(keys)) {
return keys.reduce(function(previous, key) {
let val = y.share.map.get(key);
@ -277,12 +277,12 @@ class TransportYJS extends Transport {
return typeof val === "string" ? JSON.parse(val) : val; // Surprisingly this is sync, the p_connection should have synchronised
}
}
async p_get(url, keys, {verbose=false}={}) {
return this._p_get(await this.p_connection(url, verbose), keys, {verbose});
async p_get(url, keys) {
return this._p_get(await this.p_connection(url), keys);
}
async p_delete(url, keys, {verbose=false}={}) {
let y = await this.p_connection(url, verbose);
async p_delete(url, keys) {
let y = await this.p_connection(url);
if (typeof keys === "string") {
y.share.map.delete(keys);
} else {
@ -290,22 +290,22 @@ class TransportYJS extends Transport {
}
}
async p_keys(url, {verbose=false}={}) {
let y = await this.p_connection(url, verbose);
async p_keys(url) {
let y = await this.p_connection(url);
return y.share.map.keys(); // Surprisingly this is sync, the p_connection should have synchronised
}
async p_getall(url, {verbose=false}={}) {
let y = await this.p_connection(url, verbose);
async p_getall(url) {
let y = await this.p_connection(url);
let keys = y.share.map.keys(); // Surprisingly this is sync, the p_connection should have synchronised
return this._p_get(y, keys, {verbose});
return this._p_get(y, keys);
}
async p_rawfetch(url, {verbose=false}={}) {
async p_rawfetch(url) {
return { // See identical structure in TransportHTTP
table: "keyvaluetable", //TODO-KEYVALUE its unclear if this is the best way, as maybe want to know the real type of table e.g. domain
_map: await this.p_getall(url, {verbose})
_map: await this.p_getall(url)
}; // Data struc is ok as SmartDict.p_fetch will pass to KVT constructor
}
async monitor(url, callback, {verbose=false, current=false}={}) {
async monitor(url, callback, {current=false}={}) {
/*
Setup a callback called whenever an item is added to a list, typically it would be called immediately after a p_getall to get any more items not returned by p_getall.
Stack: KVT()|KVT.p_new => KVT.monitor => (a: Transports.monitor => YJS.monitor)(b: dispatchEvent)
@ -313,7 +313,7 @@ class TransportYJS extends Transport {
:param url: string Identifier of list (as used by p_rawlist and "signedby" parameter of p_rawadd
:param callback: function({type, key, value}) Callback for each new item added to the list
:param verbose: boolean - true for debugging output
:param current: boolean - true if want events for current items on table
*/
url = typeof url === "string" ? url : url.href;
let y = this.yarrays[url];
@ -330,7 +330,7 @@ class TransportYJS extends Transport {
}
y.share.map.observe((event) => {
if (['add','update'].includes(event.type)) { // Currently ignoring deletions.
if (verbose) console.log("YJS monitor:", url, event.type, event.name, event.value);
debugyjs("YJS monitor: %o %s %s %o", url, event.type, event.name, event.value);
// ignores event.path (only in observeDeep) and event.object
if (!(event.type === "update" && event.oldValue === event.value)) {
// Dont trigger on update as seeing some loops with p_set
@ -345,16 +345,16 @@ class TransportYJS extends Transport {
})
}
static async p_test(opts={}, verbose=false) {
if (verbose) {console.log("TransportHTTP.test")}
static async p_test(opts={}) {
{console.log("TransportHTTP.test")}
try {
let transport = await this.p_setup(opts, verbose);
if (verbose) console.log("HTTP connected");
let res = await transport.p_info(verbose);
if (verbose) console.log("TransportHTTP info=",res);
res = await transport.p_status(verbose);
let transport = await this.p_setup(opts);
console.log("HTTP connected");
let res = await transport.p_info();
console.log("TransportHTTP info=",res);
res = await transport.p_status();
console.assert(res === Transport.STATUS_CONNECTED);
await transport.p_test_kvt("NACL%20VERIFY", verbose);
await transport.p_test_kvt("NACL%20VERIFY");
} catch(err) {
console.log("Exception thrown in TransportHTTP.test:", err.message);
throw err;

254
Transports.js
View File

@ -1,8 +1,8 @@
const Url = require('url');
const errors = require('./Errors');
const utils = require('./utils');
//process.env.DEBUG = "dweb-transports"; //TODO-DEBUG set at top level
const debugtransports = require('debug')('dweb-transports');
class Transports {
/*
@ -15,8 +15,8 @@ class Transports {
_optionspaused Saves paused option for setup
*/
constructor(options, verbose) {
if (verbose) console.log("Transports(%o)",options);
constructor(options) {
// THIS IS UNUSED - ALL METHODS ARE STATIC, THERE IS NO Transports INSTANCE
}
static _connected() {
@ -49,11 +49,11 @@ class Transports {
urls: Array of urls
func: Function to check support for: fetch, store, add, list, listmonitor, reverse - see supportFunctions on each Transport class
options: For future use
returns: Array of pairs of url & transport instance [ [ u1, t1], [u1, t2], [u2, t1]]
returns: Array of pairs of Url & transport instance [ [ u1, t1], [u1, t2], [u2, t1]]
throws: CodingError if urls empty or [undefined...]
*/
if (!((urls && urls[0]) || ["store", "newlisturls", "newdatabase", "newtable"].includes(func))) {
console.warn("Transports.validFor called with invalid arguments: urls=", urls, "func=", func); // FOr debugging old calling patterns with [ undefined ]
console.error("Transports.validFor called with invalid arguments: urls=", urls, "func=", func); // FOr debugging old calling patterns with [ undefined ]
return [];
}
if (!(urls && urls.length > 0)) {
@ -74,21 +74,21 @@ class Transports {
// SEE-OTHER-ADDTRANSPORT
static http(verbose) {
static http() {
// Find an http transport if it exists, so for example YJS can use it.
return Transports._connected().find((t) => t.name === "HTTP")
}
static ipfs(verbose) {
static ipfs() {
// Find an ipfs transport if it exists, so for example YJS can use it.
return Transports._connected().find((t) => t.name === "IPFS")
}
static webtorrent(verbose) {
static webtorrent() {
// Find an ipfs transport if it exists, so for example ServiceWorker.p_respondWebTorrent can use it.
return Transports._connected().find((t) => t.name === "WEBTORRENT")
}
static gun(verbose) {
static gun() {
// Find a GUN transport if it exists
return Transports._connected().find((t) => t.name === "GUN")
}
@ -109,39 +109,43 @@ class Transports {
this.namingcb = cb;
}
static async _p_rawstore(tt, data, {verbose}) {
static async _p_rawstore(tt, data) {
// Internal method to store at known transports
let errs = [];
let rr = await Promise.all(tt.map(async function(t) {
try {
return await t.p_rawstore(data, {verbose}); //url
debugtransports("Storing %d bytes to %s", data.length, t.name);
let url = await t.p_rawstore(data);
debugtransports("Storing %d bytes to %s succeeded: %s", data.length, t.name, url);
return url; //url
} catch(err) {
console.log("Could not rawstore to", t.name, err.message);
debugtransports("Storing %d bytes to %s failed: %s", data.length, t.name, err.message);
errs.push(err);
return undefined;
}
}));
rr = rr.filter((r) => !!r); // Trim any that had errors
if (!rr.length) {
debugtransports("Storing %d bytes failed on all transports", data.length);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); // New error with concatenated messages
}
return rr;
}
static async p_rawstore(data, {verbose}) {
static async p_rawstore(data) {
/*
data: Raw data to store - typically a string, but its passed on unmodified here
returns: Array of urls of where stored
throws: TransportError with message being concatenated messages of transports if NONE of them succeed.
*/
let tt = this.validFor(undefined, "store").map(([u, t]) => t); // Valid connected transports that support "store"
if (verbose) console.log("Valid for transports:", tt.map(t => t.name));
if (!tt.length) {
debugtransports("Storing %d bytes failed: no transports available", data.length);
throw new errors.TransportError('Transports.p_rawstore: Cant find transport for store');
}
return this._p_rawstore(tt, data, {verbose});
return this._p_rawstore(tt, data);
}
static async p_rawlist(urls, {verbose=false}={}) {
static async p_rawlist(urls) {
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
let tt = this.validFor(urls, "list"); // Valid connected transports that support "store"
if (!tt.length) {
@ -150,15 +154,19 @@ class Transports {
let errs = [];
let ttlines = await Promise.all(tt.map(async function([url, t]) {
try {
return await t.p_rawlist(url, {verbose}); // [sig]
debugtransports("Listing %s via %s", url, t.name);
let res = await t.p_rawlist(url); // [sig]
debugtransports("Listing %s via %s retrieved %d items", url, t.name, res.length);
return res;
} catch(err) {
console.log("Could not rawlist ", url, "from", t.name, err.message);
debugtransports("Listing %s via %s failed: %s", url, t.name, err.message);
errs.push(err);
return [];
}
})); // [[sig,sig],[sig,sig]]
if (errs.length >= tt.length) {
// All Transports failed (maybe only 1)
debugtransports("Listing %o failed on all transports", urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); // New error with concatenated messages
}
let uniques = {}; // Used to filter duplicates
@ -171,7 +179,6 @@ class Transports {
Fetch the data for a url, transports act on the data, typically storing it.
urls: array of urls to retrieve (any are valid)
opts {
verbose,
start, integer - first byte wanted
end integer - last byte wanted (note this is inclusive start=0,end=1023 is 1024 bytes
timeoutMS integer - max time to wait on transports (IPFS) that support it
@ -180,7 +187,6 @@ class Transports {
throws: TransportError with concatenated error messages if none succeed.
throws: CodingError if urls empty or [undefined ... ]
*/
let verbose = opts.verbose;
if (!urls.length) throw new errors.TransportError("Transports.p_rawfetch given an empty list of urls");
let resolvedurls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
if (!resolvedurls.length) throw new errors.TransportError("Transports.p_rawfetch none of the urls resolved: " + urls);
@ -193,27 +199,30 @@ class Transports {
let failedtransports = []; // Will accumulate any transports fail on before the success
for (const [url, t] of tt) {
try {
debugtransports("Fetching %s via %s", url.href, t.name);
let data = await t.p_rawfetch(url, opts); // throws errors if fails or timesout
debugtransports("Fetching %s via %s succeeded %d bytes", url.href, t.name, data.length);
//TODO-MULTI-GATEWAY working here - it doesnt quite work yet as the "Add" on browser gets different url than on server
if (opts.relay && failedtransports.length) {
console.log(`Relaying ${data.length} bytes from ${typeof url === "string" ? url : url.href} to ${failedtransports.map(t=>t.name)}`);
this._p_rawstore(failedtransports, data, {verbose})
.then(uu => console.log(`Relayed to ${uu}`)); // Happening async, not waiting and dont care if fails
debugtransports("Fetching attempting relay of %d bytes from %s to %o", data.length, url.href, failedtransports.map(t=>t.name));
this._p_rawstore(failedtransports, data)
.then(uu => debugtransports(`Fetching relayed %d bytes to %o`, data.length, uu)); // Happening async, not waiting and dont care if fails
}
//END TODO-MULTI-GATEWAY
return data;
} catch (err) {
failedtransports.push(t);
errs.push(err);
console.log("Could not retrieve ", url && url.href, "from", t && t.name, err.message);
debugtransports("Fetching %s via %s failed: %s", url.href, t.name, err.message);
// Don't throw anything here, loop round for next, only throw if drop out bottom
//TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
}
}
debugtransports("Fetching %o failed on all transports", urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); //Throw err with combined messages if none succeed
}
static async p_rawadd(urls, sig, {verbose=false}={}) {
static async p_rawadd(urls, sig) {
/*
urls: of lists to add to
sig: Sig to add
@ -226,20 +235,24 @@ class Transports {
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
let tt = this.validFor(urls, "add"); // Valid connected transports that support "store"
if (!tt.length) {
debugtransports("Adding to %o via %s failed: no transports available", urls, t.name);
throw new errors.TransportError('Transports.p_rawstore: Cant find transport for urls:'+urls.join(','));
}
let errs = [];
await Promise.all(tt.map(async function([u, t]) {
try {
await t.p_rawadd(u, sig, {verbose}); //undefined
debugtransports("Adding to %s via %s", u, t.name);
await t.p_rawadd(u, sig); //undefined
debugtransports("Adding to %s via %s succeeded", u, t.name);
return undefined;
} catch(err) {
console.log("Could not rawlist ", u, "from", t.name, err.message);
debugtransports("Adding to %s via %s failed: %s", u, t.name, err.message);
errs.push(err);
return undefined;
}
}));
if (errs.length >= tt.length) {
debugtransports("Adding to %o failed on all transports", urls);
// All Transports failed (maybe only 1)
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); // New error with concatenated messages
}
@ -253,53 +266,61 @@ class Transports {
*/
// Note cant do p_resolveNames since sync but should know real urls of resource by here.
this.validFor(urls, "listmonitor")
.map(([u, t]) => t.listmonitor(u, cb, opts));
.map(([u, t]) => {
t.listmonitor(u, cb, opts);
debugtransports("Monitoring list %s via %s", u, t.name);
});
}
static async p_newlisturls(cl, {verbose=false}={}) {
static async p_newlisturls(cl) {
// Create a new list in any transport layer that supports lists.
// cl is a CommonList or subclass and can be used by the Transport to get info for choosing the list URL (normally it won't use it)
// Note that normally the CL will not have been stored yet, so you can't use its urls.
let uuu = await Promise.all(this.validFor(undefined, "newlisturls")
.map(([u, t]) => t.p_newlisturls(cl, {verbose})) ); // [ [ priv, pub] [ priv, pub] [priv pub] ]
.map(([u, t]) => t.p_newlisturls(cl)) ); // [ [ priv, pub] [ priv, pub] [priv pub] ]
return [uuu.map(uu=>uu[0]), uuu.map(uu=>uu[1])]; // [[ priv priv priv ] [ pub pub pub ] ]
}
// Stream handling ===========================================
//myArray[Math.floor(Math.random() * myArray.length)];
static async p_f_createReadStream(urls, {verbose=false, wanturl=false}={}) { // Note options is options for selecting a stream, not the start/end in a createReadStream call
static async p_f_createReadStream(urls, {wanturl=false}={}) { // Note options is options for selecting a stream, not the start/end in a createReadStream call
/*
urls: Urls of the stream
returns: f(opts) => stream returning bytes from opts.start || start of file to opts.end-1 || end of file
*/
let tt = this.validFor(urls, "createReadStream", {}); //[ [Url,t],[Url,t]] // Can pass options TODO-STREAM support options in validFor
if (!tt.length) {
debugtransports("Opening stream to %o via %s failed: no transports available", urls, t.name);
throw new errors.TransportError("Transports.p_createReadStream cant find any transport for urls: " + urls);
}
//With multiple transports, it should return when the first one returns something.
let errs = [];
// Until we have transport ordering, try randomly
// Until we have transport ordering, try randomly TODO Transport ordering
let z;
while ((z = tt.splice(Math.floor(Math.random() * tt.length),1)).length) {
let url = z[0][0];
let t = z[0][1];
try {
return await t.p_f_createReadStream(url, {verbose, wanturl} );
debugtransports("Opening stream to %s via %s", url.href, t.name);
let res = await t.p_f_createReadStream(url, {wanturl} );
debugtransports("Opening stream to %s via %s succeeded", url.href, t.name);
return res;
} catch (err) {
errs.push(err);
console.log("Could not retrieve ", url.href, "from", t.name, err.message);
debugtransports("Opening stream to %s via %s failed: %s", url.href, t.name, err.message);
// Don't throw anything here, loop round for next, only throw if drop out bottom
//TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
}
}
debugtransports("Opening stream to %o failed on all transports", urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); //Throw err with combined messages if none succeed
}
// KeyValue support ===========================================
static async p_get(urls, keys, {verbose=false}={}) {
static async p_get(urls, keys) {
/*
Fetch the values for a url and one or more keys, transports act on the data, typically storing it.
urls: array of urls to retrieve (any are valid)
@ -308,76 +329,92 @@ class Transports {
throws: TransportError with concatenated error messages if none succeed.
*/
let tt = this.validFor(urls, "get"); //[ [Url,t],[Url,t]]
let debug1 = Array.isArray(keys) ? `${keys.length} keys` : keys; // "1 keys" or "foo"
if (!tt.length) {
throw new errors.TransportError("Transports.p_get cant find any transport for urls: " + urls);
debugtransports("Getting %s from %o failed: no transports available", debug1, urls);
throw new errors.TransportError("Transports.p_get cant find any transport to get keys from urls: " + urls);
}
//With multiple transports, it should return when the first one returns something.
let errs = [];
for (const [url, t] of tt) {
try {
return await t.p_get(url, keys, {verbose}); //TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
debugtransports("Getting %s from %s via %s", debug1, url.href, t.name);
let res = await t.p_get(url, keys); //TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
debugtransports("Getting %s from %s via %s succeeded length=%d", debug1, url.href, t.name, res.length);
return res;
} catch (err) {
errs.push(err);
console.log("Could not retrieve ", url.href, "from", t.name, err.message);
debugtransports("Getting %s from %s via %s failed: %s", debug1, url.href, t.name, err.message);
// Don't throw anything here, loop round for next, only throw if drop out bottom
}
}
debugtransports("Getting %s from %o failed on all transports", debug1, urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); //Throw err with combined messages if none succeed
}
static async p_set(urls, keyvalues, value, {verbose=false}={}) {
static async p_set(urls, keyvalues, value) {
/* Set a series of key/values or a single value
keyvalues: Either dict or a string
value: if kv is a string, this is the value to set
throws: TransportError with message being concatenated messages of transports if NONE of them succeed.
*/
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
let debug1 = typeof keyvalues === "object" ? `${keyvalues.length} keys` : keyvalues; // "1 keys" or "foo"
let tt = this.validFor(urls, "set"); //[ [Url,t],[Url,t]]
if (!tt.length) {
debugtransports("Setting %s on %o failed: no transports available", debug1, urls);
throw new errors.TransportError("Transports.p_set cant find any transport for urls: " + urls);
}
let errs = [];
let success = false;
await Promise.all(tt.map(async function([url, t]) {
try {
await t.p_set(url, keyvalues, value, {verbose});
debugtransports("Setting %s on %s via %s", debug1, url.href, t.name);
await t.p_set(url, keyvalues, value);
debugtransports("Setting %s on %s via %s succeeded", debug1, url.href, t.name);
success = true; // Any one success will return true
} catch(err) {
console.log("Could not rawstore to", t.name, err.message);
debugtransports("Setting %s on %s via %s failed: %s", debug1, url.href, t.name, err.message);
errs.push(err);
}
}));
if (!success) {
debugtransports("Setting %s on %o failed on all transports", debug1, urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); // New error with concatenated messages
}
}
static async p_delete(urls, keys, {verbose=false}={}) {
static async p_delete(urls, keys) {
/* Delete a key or a list of keys
kv: Either dict or a string
value: if kv is a string, this is the value to set
throws: TransportError with message being concatenated messages of transports if NONE of them succeed.
*/
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
let debug1 = Array.isArray(keys) ? `${keys.length} keys` : keys; // "1 keys" or "foo"
let tt = this.validFor(urls, "set"); //[ [Url,t],[Url,t]]
if (!tt.length) {
debugtransports("Deleting %s on %o failed: no transports available", debug1, urls);
throw new errors.TransportError("Transports.p_set cant find any transport for urls: " + urls);
}
let errs = [];
let success = false;
await Promise.all(tt.map(async function([url, t]) {
try {
await t.p_delete(url, keys, {verbose});
debugtransports("Deleting %s on %s via %s", debug1, url.href, t.name);
await t.p_delete(url, keys);
debugtransports("Deleting %s on %s via %s succeeded", debug1, url.href, t.name);
success = true; // Any one success will return true
} catch(err) {
console.log("Could not rawstore to", t.name, err.message);
debugtransports("Deleting %s on %s via %s failed: %s", debug1, url.href, t.name, err.message);
errs.push(err);
}
}));
if (!success) {
debugtransports("Deleting %s on %o failed on all transports", debug1, urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); // New error with concatenated messages
}
}
static async p_keys(urls, {verbose=false}={}) {
static async p_keys(urls) {
/*
Fetch the values for a url and one or more keys, transports act on the data, typically storing it.
urls: array of urls to retrieve (any are valid)
@ -388,23 +425,28 @@ class Transports {
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
let tt = this.validFor(urls, "keys"); //[ [Url,t],[Url,t]]
if (!tt.length) {
debugtransports("Getting all keys on %o failed: no transports available", urls);
throw new errors.TransportError("Transports.p_keys cant find any transport for urls: " + urls);
}
//With multiple transports, it should return when the first one returns something.
//With multiple transports, it should return when the first one returns something. TODO make it return the aggregate
let errs = [];
for (const [url, t] of tt) {
try {
return await t.p_keys(url, {verbose}); //TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
debugtransports("Getting all keys on %s via %s", url.href, t.name);
let res = await t.p_keys(url); //TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
debugtransports("Getting all keys on %s via %s succeeded with %d keys", url.href, t.name, res.length);
return res;
} catch (err) {
errs.push(err);
console.log("Could not retrieve keys for", url.href, "from", t.name, err.message);
debugtransports("Getting all keys on %s via %s failed: %s", url.href, t.name, err.message);
// Don't throw anything here, loop round for next, only throw if drop out bottom
}
}
debugtransports("Getting all keys on %o failed on all transports", urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); //Throw err with combined messages if none succeed
}
static async p_getall(urls, {verbose=false}={}) {
static async p_getall(urls) {
/*
Fetch the values for a url and one or more keys, transports act on the data, typically storing it.
urls: array of urls to retrieve (any are valid)
@ -415,64 +457,73 @@ class Transports {
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
let tt = this.validFor(urls, "getall"); //[ [Url,t],[Url,t]]
if (!tt.length) {
debugtransports("Getting all values on %o failed: no transports available", urls);
throw new errors.TransportError("Transports.p_getall cant find any transport for urls: " + urls);
}
//With multiple transports, it should return when the first one returns something.
let errs = [];
for (const [url, t] of tt) {
try {
return await t.p_getall(url, {verbose}); //TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
debugtransports("Getting all values on %s via %s", url.href, t.name);
let res = await t.p_getall(url); //TODO-MULTI-GATEWAY potentially copy from success to failed URLs.
debugtransports("Getting all values on %s via %s succeeded with %d values", url.href, t.name, res.length);
return res;
} catch (err) {
errs.push(err);
console.log("Could not retrieve all keys for", url.href, "from", t.name, err.message);
debugtransports("Getting all values on %s via %s failed: %s", url.href, t.name, err.message);
// Don't throw anything here, loop round for next, only throw if drop out bottom
}
}
debugtransports("Getting all keys on %o failed on all transports", urls);
throw new errors.TransportError(errs.map((err)=>err.message).join(', ')); //Throw err with combined messages if none succeed
}
static async p_newdatabase(pubkey, {verbose=false}={}) {
static async p_newdatabase(pubkey) {
/*
Create a new database in any transport layer that supports databases (key value pairs).
pubkey: CommonList, KeyPair, or exported public key
resolves to: [ privateurl, publicurl]
*/
let uuu = await Promise.all(this.validFor(undefined, "newdatabase")
.map(([u, t]) => t.p_newdatabase(pubkey, {verbose})) ); // [ { privateurl, publicurl} { privateurl, publicurl} { privateurl, publicurl} ]
.map(([u, t]) => t.p_newdatabase(pubkey)) ); // [ { privateurl, publicurl} { privateurl, publicurl} { privateurl, publicurl} ]
return { privateurls: uuu.map(uu=>uu.privateurl), publicurls: uuu.map(uu=>uu.publicurl) }; // { privateurls: [], publicurls: [] }
}
static async p_newtable(pubkey, table, {verbose=false}={}) {
static async p_newtable(pubkey, table) {
/*
Create a new table in any transport layer that supports the function (key value pairs).
pubkey: CommonList, KeyPair, or exported public key
resolves to: [ privateurl, publicurl]
*/
let uuu = await Promise.all(this.validFor(undefined, "newtable")
.map(([u, t]) => t.p_newtable(pubkey, table, {verbose})) ); // [ [ priv, pub] [ priv, pub] [priv pub] ]
.map(([u, t]) => t.p_newtable(pubkey, table)) ); // [ [ priv, pub] [ priv, pub] [priv pub] ]
return { privateurls: uuu.map(uu=>uu.privateurl), publicurls: uuu.map(uu=>uu.publicurl)}; // {privateurls: [ priv priv priv ], publicurls: [ pub pub pub ] }
}
static async p_connection(urls, verbose) {
static async p_connection(urls) {
/*
Do any asynchronous connection opening work prior to potentially synchronous methods (like monitor)
*/
urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
await Promise.all(
this.validFor(urls, "connection")
.map(([u, t]) => t.p_connection(u, verbose)));
.map(([u, t]) => t.p_connection(u)));
}
static monitor(urls, cb, {verbose=false, current=false}={}) {
static monitor(urls, cb, {current=false}={}) {
/*
Add a listmonitor for each transport - note this means if multiple transports support it, then will get duplicate events back if everyone else is notifying all of them.
Stack: KVT()|KVT.p_new => KVT.monitor => (a: Transports.monitor => YJS.monitor)(b: dispatchEvent)
cb: function({type, key, value})
current: If true then then send all current entries as well
*/
//Cant' its async. urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
//Can't its async. urls = await this.p_resolveNames(urls); // If naming is loaded then convert to a name
this.validFor(urls, "monitor")
.map(([u, t]) => t.monitor(u, cb, {verbose, current}));
.map(([u, t]) => {
debugtransports("Monitoring table %s via %s", u, t.name);
t.monitor(u, cb, {current})
}
);
}
// Setup and connection
@ -486,10 +537,11 @@ class Transports {
// Setup Transports - setup0 is called once, and should return quickly, p_setup1 and p_setup2 are asynchronous and p_setup2 relies on p_setup1 having resolved.
static setup0(tabbrevs, options, verbose, cb) {
static setup0(tabbrevs, options, cb) {
/*
Setup Transports for a range of classes
tabbrevs is abbreviation HTTP, IPFS, LOCAL or list of them e.g. "HTTP,IPFS"
cb is callback for when status changes, but there are no status changes here so its not called.
Handles "LOCAL" specially, turning into a HTTP to a local server (for debugging)
returns array of transport instances
@ -497,34 +549,35 @@ class Transports {
// "IPFS" or "IPFS,LOCAL,HTTP"
let localoptions = {http: {urlbase: "http://localhost:4244"}};
return tabbrevs.map((tabbrev) => {
let transportclass;
if (tabbrev === "LOCAL") {
transportclass = this._transportclasses["HTTP"];
} else {
transportclass = this._transportclasses[tabbrev];
}
let transportclass = this._transportclasses[ (tabbrev === "LOCAL") ? "HTTP" : tabbrev ];
if (!transportclass) {
let tt = Object.keys(this._transportclasses);
console.error(`Requested ${tabbrev} but ${tt.length ? tt : "No"} transports have been loaded`);
debugtransports("Connection to %s unavailable", tabbrev);
return undefined;
} else {
return transportclass.setup0(tabbrev === "LOCAL" ? localoptions : options, verbose);
debugtransports("Setting up connection to %s with options %o", tabbrev, options);
return transportclass.setup0(tabbrev === "LOCAL" ? localoptions : options);
}
}).filter(f => !!f); // Trim out any undefined
}
static async p_setup1(verbose, cb) {
static async p_setup1(cb) {
/* Second stage of setup, connect if possible */
// Does all setup1a before setup1b since 1b can rely on ones with 1a, e.g. YJS relies on IPFS
await Promise.all(this._transports
.filter((t) => (! this._optionspaused.includes(t.name)))
.map((t) => t.p_setup1(verbose, cb)))
.map((t) => {
debugtransports("Connection stage 1 to %s", t.name);
return t.p_setup1(cb);
}))
}
static async p_setup2(verbose, cb) {
static async p_setup2(cb) {
/* Second stage of setup, connect if possible */
// Does all setup1a before setup1b since 1b can rely on ones with 1a, e.g. YJS relies on IPFS
await Promise.all(this._transports
.filter((t) => (! this._optionspaused.includes(t.name)))
.map((t) => t.p_setup2(verbose, cb)))
.map((t) => {
debugtransports("Connection stage 2 to %s", t.name);
return t.p_setup2(cb);
}))
}
static async refreshstatus(t) {
@ -540,22 +593,20 @@ class Transports {
}
}
static async p_connect(options, verbose) {
static async p_connect(options) {
/*
This is a standardish starting process, feel free to subclass or replace !
It will connect to a set of standard transports and is intended to work inside a browser.
options = { defaulttransports: ["IPFS"], statuselement: el, http: {}, ipfs: {}; paused: ["IPFS"] }
*/
if (verbose) console.group("p_connect ---");
try {
options = options || {};
let setupoptions = {};
let tabbrevs = options.transports; // Array of transport abbreviations
this._optionspaused = (options.paused || []).map(n => n.toUpperCase()); // Array of transports paused - defaults to none, upper cased
if (!(tabbrevs && tabbrevs.length)) { tabbrevs = options.defaulttransports || [] }
if (! tabbrevs.length) { tabbrevs = ["HTTP", "YJS", "IPFS", "WEBTORRENT", "GUN"]; } // SEE-OTHER-ADDTRANSPORT
tabbrevs = tabbrevs.map(n => n.toUpperCase());
let transports = this.setup0(tabbrevs, options, verbose);
let transports = this.setup0(tabbrevs, options);
if (options.statuscb) {
this.statuscb = options.statuscb;
}
@ -572,13 +623,12 @@ class Transports {
}
)));
}
await this.p_setup1(verbose, this.refreshstatus);
await this.p_setup2(verbose, this.refreshstatus);
await this.p_setup1(this.refreshstatus);
await this.p_setup2(this.refreshstatus);
} catch(err) {
console.error("ERROR in p_connect:",err.message);
throw(err);
}
if (verbose) console.groupEnd("p_connect ---");
}
static async p_urlsFrom(url) {
@ -606,43 +656,43 @@ class Transports {
return Transports.http()._url(urls.find(u => (u.startsWith("contenthash") || u.startsWith("http") )), "content/rawfetch");
}
static async test(verbose) {
if (verbose) {console.log("Transports.test")}
static async test() {
console.log("Transports.test")
try {
/* Could convert this - copied fom YJS to do a test at the "Transports" level
let testurl = "yjs:/yjs/THISATEST"; // Just a predictable number can work with
let res = await transport.p_rawlist(testurl, verbose);
let res = await transport.p_rawlist(testurl);
let listlen = res.length; // Holds length of list run intermediate
if (verbose) console.log("rawlist returned ", ...utils.consolearr(res));
transport.listmonitor(testurl, (obj) => console.log("Monitored", obj), verbose);
let sig = new Dweb.Signature({urls: ["123"], date: new Date(Date.now()), signature: "Joe Smith", signedby: [testurl]}, verbose);
await transport.p_rawadd(testurl, sig, {verbose});
if (verbose) console.log("TransportIPFS.p_rawadd returned ");
res = await transport.p_rawlist(testurl, verbose);
if (verbose) console.log("rawlist returned ", ...utils.consolearr(res)); // Note not showing return
console.log("rawlist returned ", ...utils.consolearr(res));
transport.listmonitor(testurl, (obj) => console.log("Monitored", obj));
let sig = new Dweb.Signature({urls: ["123"], date: new Date(Date.now()), signature: "Joe Smith", signedby: [testurl]});
await transport.p_rawadd(testurl, sig);
console.log("TransportIPFS.p_rawadd returned ");
res = await transport.p_rawlist(testurl);
console.log("rawlist returned ", ...utils.consolearr(res)); // Note not showing return
await delay(500);
res = await transport.p_rawlist(testurl, verbose);
res = await transport.p_rawlist(testurl);
console.assert(res.length === listlen + 1, "Should have added one item");
*/
//console.log("TransportYJS test complete");
/* TODO-KEYVALUE reenable these tests,s but catch http examples
let db = await this.p_newdatabase("TESTNOTREALLYAKEY", {verbose}); // { privateurls, publicurls }
let db = await this.p_newdatabase("TESTNOTREALLYAKEY"); // { privateurls, publicurls }
console.assert(db.privateurls[0] === "yjs:/yjs/TESTNOTREALLYAKEY");
let table = await this.p_newtable("TESTNOTREALLYAKEY","TESTTABLE", {verbose}); // { privateurls, publicurls }
let table = await this.p_newtable("TESTNOTREALLYAKEY","TESTTABLE"); // { privateurls, publicurls }
let mapurls = table.publicurls;
console.assert(mapurls[0] === "yjs:/yjs/TESTNOTREALLYAKEY/TESTTABLE");
await this.p_set(mapurls, "testkey", "testvalue", {verbose});
let res = await this.p_get(mapurls, "testkey", {verbose});
await this.p_set(mapurls, "testkey", "testvalue");
let res = await this.p_get(mapurls, "testkey");
console.assert(res === "testvalue");
await this.p_set(mapurls, "testkey2", {foo: "bar"}, {verbose});
res = await this.p_get(mapurls, "testkey2", {verbose});
await this.p_set(mapurls, "testkey2", {foo: "bar"});
res = await this.p_get(mapurls, "testkey2");
console.assert(res.foo === "bar");
await this.p_set(mapurls, "testkey3", [1,2,3], {verbose});
res = await this.p_get(mapurls, "testkey3", {verbose});
await this.p_set(mapurls, "testkey3", [1,2,3]);
res = await this.p_get(mapurls, "testkey3");
console.assert(res[1] === 2);
res = await this.p_keys(mapurls);
console.assert(res.length === 3 && res.includes("testkey3"));
res = await this.p_getall(mapurls, {verbose});
res = await this.p_getall(mapurls);
console.assert(res.testkey2.foo === "bar");
*/

6456
dist/dweb-transports-bundle.js vendored
View File

File diff suppressed because one or more lines are too long

9
example_block.html
View File

@ -5,6 +5,7 @@
<!--Note there are copies of this in dweb-transports repo and dweb-transport repo-->
<meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
<title>DWEB Transports example page</title>
<script type="text/javascript">localStorage.debug = "dweb-transports dweb-transports:*";</script>
<script type="text/javascript" src="./dist/dweb-transports-bundle.js"></script>
<script type="text/javascript" src='https://cloud.tinymce.com/stable/tinymce.min.js'></script><!-- TinyMCE -->
<script type="text/javascript">
@ -18,11 +19,10 @@
'insertdatetime media table contextmenu paste code' ],
toolbar: 'save | undo redo | insert | styleselect | bold italic | alignleft aligncenter alignright alignjustify | bullist numlist outdent indent | link image',
save_onsavecallback: function() {
let verbose = true; // See commentary in console.log
let content = tinyMCE.get('mytextarea').getContent();
// alert(content);
let el = document.getElementById("retrievalarea");
let urls = DwebTransports.p_rawstore(content, {verbose})
let urls = DwebTransports.p_rawstore(content)
.then((urls) => el.value = urls)
.catch((err) => {console.error("saveoncallback", err); alert(err)});
}
@ -40,7 +40,7 @@
let destn = document.getElementById("retrievaldestn");
destn.innerHTML = ""; // Clear it first
let data = await DwebTransports.p_rawfetch(urls, {verbose});
let data = await DwebTransports.p_rawfetch(urls);
console.log("Retrieved data length", data.length);
destn.innerHTML = data;
} catch(err) {
@ -56,7 +56,7 @@
statuselement: document.getElementById("statuselement"), // Where to build status indicator
defaulttransports: ["HTTP","IPFS"], // Default transports if not specified
transports: searchparams.getAll("transport") // Allow override default from URL parameters
}, verbose);
});
// Any code you want to run after connected to transports goes here.
if (url) fetchanddisplay(url);
} catch(err) {
@ -114,7 +114,6 @@
<script type="text/javascript">
var searchparams = new URL(window.location.href).searchParams;
var verbose = searchparams.get("verbose") || false; // Anything specified for verbose is true (could truthify buy no need)
main(searchparams.get("url")); //starts a transport, checks its status, and if appropriate to app opens a URL passed
</script>

21
httptools.js
View File

@ -1,5 +1,6 @@
const nodefetch = require('node-fetch'); // Note, were using node-fetch-npm which had a warning in webpack see https://github.com/bitinn/node-fetch/issues/421 and is intended for clients
const errors = require('./Errors'); // Standard Dweb Errors
const debught = require('debug')('dweb-transports:httptools');
//var fetch,Headers,Request;
//if (typeof(Window) === "undefined") {
@ -38,21 +39,21 @@ async function loopfetch(req, ms, count, what) {
return await fetch(req);
} catch(err) {
lasterr = err;
console.log("Delaying", what,"by", ms, "because", err.message);
debught("Delaying %s by %d ms because %s", what, ms, err.message);
await new Promise(resolve => {setTimeout(() => { resolve(); },ms)})
ms = ms*(1+Math.random()); // Spread out delays incase all requesting same time
}
}
console.log("Looping",what,"failed");
console.warn("loopfetch of",what,"failed");
if (loopguard !== (window && window.loopguard)) {
console.log("Looping exited because of page change "+ what);
debught("Looping exited because of page change %s", what);
throw new Error("Looping exited because of page change "+ what)
} else {
throw(lasterr);
}
}
httptools.p_httpfetch = async function(httpurl, init, {verbose=false, wantstream=false}={}) { // Embrace and extend "fetch" to check result etc.
httptools.p_httpfetch = async function(httpurl, init, {wantstream=false}={}) { // Embrace and extend "fetch" to check result etc.
/*
Fetch a url
@ -62,7 +63,7 @@ httptools.p_httpfetch = async function(httpurl, init, {verbose=false, wantstream
throws: TransportError if fails to fetch
*/
try {
if (verbose) console.log("p_httpfetch:", httpurl, JSON.stringify(init));
debught("p_httpfetch: %s %o", httpurl, init);
//console.log('CTX=',init["headers"].get('Content-Type'))
// Using window.fetch, because it doesn't appear to be in scope otherwise in the browser.
let req = new Request(httpurl, init);
@ -86,7 +87,7 @@ httptools.p_httpfetch = async function(httpurl, init, {verbose=false, wantstream
throw new errors.TransportError(`Transport Error ${response.status}: ${response.statusText}`);
} catch (err) {
// Error here is particularly unhelpful - if rejected during the COrs process it throws a TypeError
console.log("Note error from fetch might be misleading especially TypeError can be Cors issue:",httpurl);
debught("p_httpfetch failed but error of %s could be misleading since TypeErrors are generated by CORS or the Chrome anti DDOS 'feature'", err.message);
if (err instanceof errors.TransportError) {
throw err;
} else {
@ -102,7 +103,7 @@ httptools.p_GET = async function(httpurl, opts={}) {
opts {
start, end, // Range of bytes wanted - inclusive i.e. 0,1023 is 1024 bytes
wantstream, // Return a stream rather than data
verbose }
}
resolves to: URL that can be used to fetch the resource, of form contenthash:/contenthash/Q123
*/
let headers = new Headers();
@ -115,9 +116,9 @@ httptools.p_GET = async function(httpurl, opts={}) {
redirect: 'follow', // Chrome defaults to manual
keepalive: true // Keep alive - mostly we'll be going back to same places a lot
};
return await httptools.p_httpfetch(httpurl, init, {verbose: opts.verbose, wantstream: opts.wantstream}); // This s a real http url
return await httptools.p_httpfetch(httpurl, init, {wantstream: opts.wantstream}); // This s a real http url
}
httptools.p_POST = async function(httpurl, type, data, verbose) {
httptools.p_POST = async function(httpurl, type, data) {
// Locate and return a block, based on its url
// Throws TransportError if fails
//let headers = new window.Headers();
@ -134,7 +135,7 @@ httptools.p_POST = async function(httpurl, type, data, verbose) {
redirect: 'follow', // Chrome defaults to manual
keepalive: true // Keep alive - mostly we'll be going back to same places a lot
};
return await httptools.p_httpfetch(httpurl, init, {verbose});
return await httptools.p_httpfetch(httpurl, init);
}
exports = module.exports = httptools;

2
package.json
View File

@ -49,5 +49,5 @@
"test": "cd src; node ./test.js",
"help": "echo 'test (test it)'; echo 'build (creates dweb-transports-bundle)'"
},
"version": "0.1.8"
"version": "0.1.9"
}

6
test.js
View File

@ -1,13 +1,13 @@
const DwebTransports = require('./index.js');
async function p_test({verbose=true, transport=["GUN"]}={}) {
async function p_test({transport=["GUN"]}={}) {
if (Array.isArray(transport)) {
for (tname of transport) {
await p_test({verbose, transport: tname}); // Note this is going to run in parallel
await p_test({ transport: tname}); // Note this is going to run in parallel
}
} else {
let tclass = DwebTransports._transportclasses[transport];
await tclass.p_test({verbose});
await tclass.p_test();
}
}
p_test();

2
test_ipfs.js
View File

@ -54,7 +54,7 @@ function multihashFrom(url) {
throw new errors.CodingError(`Cant turn ${url} into a multihash`);
}
function p_ipfsstart(verbose) {
function p_ipfsstart() {
return new Promise((resolve, reject) => {
if (!usehttpapi) {
ipfs = new IPFS(defaultipfsoptions);