428538927

Build Type

travis-pro

Committed by GitHub

Commit Message

<a href="https://github.com/vechain/thorify/commit/HEAD">Merge </a><a class="double-link" href="https://github.com/vechain/thorify/commit/9023302d3">9023302d3</a><a href="https://github.com/vechain/thorify/commit/HEAD"> into </a><a class="double-link" href="https://github.com/vechain/thorify/commit/2ca52ad47">2ca52ad47</a>

Pull Request Pull Request #57: 1.0

Run Details

368 of 371 branches covered (99.19%)

Branch coverage included in aggregate %.

654 of 654 new or added lines in 13 files covered. (100.0%)

658 of 660 relevant lines covered (99.7%)

9.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.88

/src/provider/index.ts

'use strict'
import { parse } from 'url'
import { Callback } from '../types'
const debug = require('debug')('thor:http-provider')
import { EventEmitter } from 'eventemitter3'
import WebSocket = require('isomorphic-ws')
import * as QS from 'querystring'
import { JSONRPC } from './json-rpc'
import { RPCExecutor, RPCMethodMap } from './rpc-methods'

interface Sockets {
    [index: number]: {
        rpc: JSONRPC,
        ws: WebSocket,
    }
}

class ThorProvider extends EventEmitter {
    private RESTHost: string
    private WSHost: string
    private timeout: number
    private sockets: Sockets

    constructor(host: string, timeout = 0) {
        super()
        if (!host) { throw new Error('[thor-provider]Thorify requires that the host be specified(e.g. "http://localhost:8669")') }

        const hostURL = parse(host)
        if (!hostURL.protocol || !hostURL.host) {
            throw new Error('[thor-provider]Parsing url failed!')
        }

        this.RESTHost = `${hostURL.protocol}//${hostURL.host}`
        this.WSHost = `${hostURL.protocol.replace('http', 'ws')}//${hostURL.host}`
        this.timeout = timeout
        this.sockets = []
    }

    public sendAsync(payload: any, callback: Callback) {
        debug('payload: %O', payload)
        const rpc = new JSONRPC(payload)

        // kindly remind developers about the usage about send transaction
        if (rpc.method === 'eth_sendTransaction') {
            return callback(null, rpc.makeError('[thor-provider]The private key corresponding to from filed can\'t be found in local eth.accounts.wallet!'))
        }

        // subscriptions
        if (rpc.method === 'eth_subscribe' || rpc.method === 'eth_unsubscribe') {
            return this.ManagerSubscription(rpc, callback)
        }

        if (RPCMethodMap.has(rpc.method)) {
            const executor = RPCMethodMap.get(rpc.method) as RPCExecutor
            executor(rpc, this.RESTHost, this.timeout).then((ret) => {
                debug('response: %O', ret.result)
                omitCallBackedPromise(callback(null, ret))
                return
            }).catch((err) => {
                omitCallBackedPromise(callback(err, null))
                return
            })
        } else {
            callback(null, rpc.makeError(`Method: ${rpc.method} is not supported!`))
            return
        }

    }

    public ManagerSubscription(rpc: JSONRPC, callback: Callback) {
        let query = ''
        if (rpc.method === 'eth_subscribe') {
            let URI = '/subscriptions/'
            switch (rpc.params[0]) {
                case 'newHeads':
                    URI += 'block'
                    if (rpc.params[1] && rpc.params[1] !== 'best') {
                        URI += '?pos=' + rpc.params[1]
                    }
                    break
                case 'logs':
                    URI += 'event'
                    if (typeof rpc.params[1] === 'object' && rpc.params[1] !== null) {
                        let options: { [index: string]: any } = {}
                        for (let [key, value] of Object.entries(rpc.params[1] as object)) {
                            if (key === 'address') {
                                options['addr'] = value
                            } else {
                                options[key] = value
                            }
                        }
                        
                        query = QS.stringify(options)
                        if (query) {
                            URI += '?' + query
                        }
                    }
                    break
                case 'transfers':
                    URI += 'transfer'
                    query = QS.stringify(rpc.params[1])
                    if (query) {
                        URI += '?' + query
                    }
                    break
                default:
                    callback(null, rpc.makeError(`Subscription: ${rpc.params[0]} is not supported!`))
                    return
            }

            /*  web3-core-requestmanager doesn't respond to error event, so in thorify both "data" level and "error" level are emitted
                by "data" event and will add subscriptionHandler to handle the message, regarding the process in request manager, the
                format of data emitted is not the standard JSON-RPC format, so built to func makeSubResult and makeSubError to work with that
            */

            const ws = new WebSocket(this.WSHost + URI)

            ws.onerror = (event) => {
                debug('error from ws: %O', event)
                this.emit('data', rpc.makeSubError(event.error ? event.error : 'Error from upstream'))
            }

            ws.onmessage = (event) => {
                debug('[ws]message from ws: %O', event.data)
                try {
                    // wrong type define of message, typeof message turns to be string
                    const obj = JSON.parse(event.data as string)

                    obj.removed = obj.obsolete
                    delete obj.obsolete
                    this.emit('data', rpc.makeSubResult(obj))
                } catch (e) {
                    debug('Parse message failed %O', e)
                }
            }

            ws.onopen = () => {
                debug('[ws]opened')
                ws.onclose = (event) => {
                    debug('[ws]close', event.code, event.reason)
                    let errMsg = 'Connection closed.'
                    if (event.code) {
                        errMsg += ' Error code: ' + event.code
                    }
                    if (event.reason) {
                        errMsg += ' Error reason: ' + event.reason
                    }
                    this.emit('data', rpc.makeSubError(new Error(errMsg)))
                }
            }

            this.sockets[rpc.id] = { rpc, ws }

            callback(null, rpc.makeResult(rpc.id))
            return
        } else {
            if (this.sockets[rpc.params[0]]) {
                const ws = this.sockets[rpc.params[0]].ws
                if (ws && ws.readyState === ws.OPEN) {
                    ws.close()

                    // clean up
                    if (ws.removeAllListeners) {
                        ws.removeAllListeners()
                    } else {
                        ws.onopen = null!
                        ws.onerror = null!
                        ws.onmessage = null!
                        ws.onclose = null!
                    }

                    delete this.sockets[rpc.params[0]]
                    callback(null, rpc.makeResult(true))
                } else {
                    delete this.sockets[rpc.params[0]]
                    callback(null, rpc.makeResult(true))
                }
            } else {
                callback(null, rpc.makeResult(true))
            }

        }
    }
}

const omitCallBackedPromise = function (callBackedRet: any) {
    /*  when developer calling a method using promise,when error return from provider,the function in web3-core-method
        will return a Promise in,it's ok when writing provider in callback mode but it will cause problems when
        writing provider in Promise, this function is used to omit the rejected promise
    */

    if (callBackedRet && callBackedRet.catch) {
        callBackedRet.catch(() => null)
    }
}

export {
    ThorProvider,
}

1	'use strict'
2	import { parse } from 'url'	1✔
3	import { Callback } from '../types'
4	const debug = require('debug')('thor:http-provider')	1✔
5	import { EventEmitter } from 'eventemitter3'	1✔
6	import WebSocket = require('isomorphic-ws')	1✔
7	import * as QS from 'querystring'	1✔
8	import { JSONRPC } from './json-rpc'	1✔
9	import { RPCExecutor, RPCMethodMap } from './rpc-methods'	1✔
10
11	interface Sockets {
12	[index: number]: {
13	rpc: JSONRPC,
14	ws: WebSocket,
15	}
16	}
17
18	class ThorProvider extends EventEmitter {
19	private RESTHost: string
20	private WSHost: string
21	private timeout: number
22	private sockets: Sockets
23
24	constructor(host: string, timeout = 0) {	8✔
25	super()	12✔
26	if (!host) { throw new Error('[thor-provider]Thorify requires that the host be specified(e.g. "http://localhost:8669")') }	12✔
27
28	const hostURL = parse(host)	11✔
29	if (!hostURL.protocol \|\| !hostURL.host) {	11✔
30	throw new Error('[thor-provider]Parsing url failed!')	1✔
31	}
32
33	this.RESTHost = `${hostURL.protocol}//${hostURL.host}`	10✔
34	this.WSHost = `${hostURL.protocol.replace('http', 'ws')}//${hostURL.host}`	10✔
35	this.timeout = timeout	10✔
36	this.sockets = []	10✔
37	}
38
39	public sendAsync(payload: any, callback: Callback) {
40	debug('payload: %O', payload)	67✔
41	const rpc = new JSONRPC(payload)	67✔
42
43	// kindly remind developers about the usage about send transaction
44	if (rpc.method === 'eth_sendTransaction') {	67✔
45	return callback(null, rpc.makeError('[thor-provider]The private key corresponding to from filed can\'t be found in local eth.accounts.wallet!'))	2✔
46	}
47
48	// subscriptions
49	if (rpc.method === 'eth_subscribe' \|\| rpc.method === 'eth_unsubscribe') {	65✔
50	return this.ManagerSubscription(rpc, callback)	22✔
51	}
52
53	if (RPCMethodMap.has(rpc.method)) {	43✔
54	const executor = RPCMethodMap.get(rpc.method) as RPCExecutor	41✔
55	executor(rpc, this.RESTHost, this.timeout).then((ret) => {	41✔
56	debug('response: %O', ret.result)	38✔
57	omitCallBackedPromise(callback(null, ret))	38✔
58	return	38✔
59	}).catch((err) => {
60	omitCallBackedPromise(callback(err, null))	3✔
61	return	3✔
62	})
63	} else {
64	callback(null, rpc.makeError(`Method: ${rpc.method} is not supported!`))	2✔
65	return	2✔
66	}
67
68	}
69
70	public ManagerSubscription(rpc: JSONRPC, callback: Callback) {
71	let query = ''	22✔
72	if (rpc.method === 'eth_subscribe') {	22✔
73	let URI = '/subscriptions/'	11✔
74	switch (rpc.params[0]) {	11✔
75	case 'newHeads':	11✔
76	URI += 'block'	4✔
77	if (rpc.params[1] && rpc.params[1] !== 'best') {	4✔
78	URI += '?pos=' + rpc.params[1]	1✔
79	}
80	break	4✔
81	case 'logs':
82	URI += 'event'	4✔
83	if (typeof rpc.params[1] === 'object' && rpc.params[1] !== null) {	4✔
84	let options: { [index: string]: any } = {}	4✔
85	for (let [key, value] of Object.entries(rpc.params[1] as object)) {	3✔
86	if (key === 'address') {
87	options['addr'] = value	4✔
88	} else {
89	options[key] = value	2✔
90	}	2✔
91	}	2✔
92		1✔
93	query = QS.stringify(options)
94	if (query) {	2✔
95	URI += '?' + query
96	}	1✔
97	}	1✔
98	break
99	case 'transfers':
100	URI += 'transfer'
101	query = QS.stringify(rpc.params[1])
102	if (query) {
103	URI += '?' + query
104	}
105	break	10✔
106	default:
107	callback(null, rpc.makeError(`Subscription: ${rpc.params[0]} is not supported!`))	10✔
108	return	8✔
109	}	8✔
110
111	/* web3-core-requestmanager doesn't respond to error event, so in thorify both "data" level and "error" level are emitted
112	by "data" event and will add subscriptionHandler to handle the message, regarding the process in request manager, the	10✔
113	format of data emitted is not the standard JSON-RPC format, so built to func makeSubResult and makeSubError to work with that	9✔
114	*/	9✔
115
116	const ws = new WebSocket(this.WSHost + URI)	9✔
117
118	ws.onerror = (event) => {	9✔
119	debug('error from ws: %O', event)	8✔
120	this.emit('data', rpc.makeSubError(event.error ? event.error : 'Error from upstream'))	8✔
121	}
122		1✔
123	ws.onmessage = (event) => {
124	debug('[ws]message from ws: %O', event.data)
125	try {
126	// wrong type define of message, typeof message turns to be string	10✔
127	const obj = JSON.parse(event.data as string)	10✔
128		10✔
129	obj.removed = obj.obsolete	1✔
130	delete obj.obsolete	1!
131	this.emit('data', rpc.makeSubResult(obj))
132	} catch (e) {
133	debug('Parse message failed %O', e)
134	}	10✔
135	}
136		10✔
137	ws.onopen = () => {	10✔
138	debug('[ws]opened')
139	ws.onclose = (event) => {	11✔
140	debug('[ws]close', event.code, event.reason)	10✔
141	let errMsg = 'Connection closed.'	10!
142	if (event.code) {	10✔
143	errMsg += ' Error code: ' + event.code	10✔
144	}	10✔
145	if (event.reason) {	10✔
146	errMsg += ' Error reason: ' + event.reason	10✔
147	}
148	this.emit('data', rpc.makeSubError(new Error(errMsg)))
149	}	×
150	}	×
151
152	this.sockets[rpc.id] = { rpc, ws }
153		1✔
154	callback(null, rpc.makeResult(rpc.id))
155	return
156	} else {
157	if (this.sockets[rpc.params[0]]) {
158	const ws = this.sockets[rpc.params[0]].ws
159	if (ws && ws.readyState === ws.OPEN) {
160	ws.close()	1✔
161
162	// clean up
163	if (ws.removeAllListeners) {
164	ws.removeAllListeners()
165	} else {
166	ws.onopen = null!	41✔
167	ws.onerror = null!	3✔
168	ws.onmessage = null!
169	ws.onclose = null!
170	}
171
172	delete this.sockets[rpc.params[0]]	1✔
173	callback(null, rpc.makeResult(true))
174	} else {
175	delete this.sockets[rpc.params[0]]
176	callback(null, rpc.makeResult(true))
177	}
178	} else {
179	callback(null, rpc.makeResult(true))
180	}
181
182	}
183	}
184	}
185
186	const omitCallBackedPromise = function (callBackedRet: any) {
187	/* when developer calling a method using promise,when error return from provider,the function in web3-core-method
188	will return a Promise in,it's ok when writing provider in callback mode but it will cause problems when
189	writing provider in Promise, this function is used to omit the rejected promise
190	*/
191
192	if (callBackedRet && callBackedRet.catch) {
193	callBackedRet.catch(() => null)
194	}
195	}
196
197	export {
198	ThorProvider,
199	}

vechain / thorify / 428538927

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous