Skip to content
Permalink
Browse files

feat(EventListener): all methods take contract implementation object

  • Loading branch information...
satello committed Apr 5, 2018
1 parent 5ffd7af commit 4a89382d03c86739bfb4f7d43f014e7ca3adc0ee
Showing with 89 additions and 92 deletions.
  1. +1 −3 src/kleros.js
  2. +2 −4 src/resources/Disputes.js
  3. +1 −1 src/resources/Notifications.js
  4. +75 −58 src/utils/EventListener.js
  5. +10 −26 tests/integration/EventListener.test.js
@@ -90,9 +90,7 @@ class Kleros {
this.eventListener.stopWatchingForEvents()
}
// reinitialize with current arbitrator contract instance
this.eventListener = new EventListener([
this.arbitrator.getContractInstance()
])
this.eventListener = new EventListener([this.arbitrator])
// add handlers for notifications
this.notifications.registerArbitratorNotifications(
account,
@@ -57,10 +57,8 @@ class Disputes {
for (let event in eventHandlerMap) {
if (eventHandlerMap.hasOwnProperty(event)) {
eventHandlerMap[event].forEach(handler => {
eventListener.addEventHandler(
this._ArbitratorInstance.getContractAddress(),
event,
args => handler(args, account)
eventListener.addEventHandler(this._ArbitratorInstance, event, args =>
handler(args, account)
)
})
}
@@ -68,7 +68,7 @@ class Notifications {
for (let event in eventHandlerMap) {
if (eventHandlerMap.hasOwnProperty(event)) {
eventListener.addEventHandler(
this._ArbitratorInstance.getContractAddress(),
this._ArbitratorInstance,
event,
this._createHandler(eventHandlerMap[event], account, callback)
)
@@ -7,9 +7,9 @@ import * as errorConstants from '../constants/error'
class EventListener {
/**
* Listen for events in contract and handles callbacks with registered event handlers.
* @param {object[]} _contractInstances - truffle-contract instances to fetch event logs for.
* @param {object[]} _contractImplementations - Contract Implementation instances to fetch event logs for.
*/
constructor(_contractInstances = []) {
constructor(_contractImplementations = []) {
this.contractInstances = []
// map address -> { event: [handlers], ... }
this.contractEventHandlerMap = {}
@@ -18,25 +18,27 @@ class EventListener {
// event handler queue
this.eventHandlerQueue = new PromiseQueue()
// initialize class variables for new contract instances
_contractInstances.forEach(instance => {
this.addContractInstance(instance)
_contractImplementations.forEach(instance => {
this.addContractImplementation(instance)
})
}

/**
* Fetch all logs from contractInstance in range.
* @param {object} contractInstance - truffle-contract instance.
* @param {object} contractImplementationInstance - Contract Implementation instance.
* @param {number} firstBlock - Lower bound of search range.
* @param {number} lastBlock - Upper bound of search range.
* @returns {Promise} All events in block range.
*/
static getAllEventLogs = (
contractInstance = isRequired('contractInstance'),
static getAllEventLogs = async (
contractImplementationInstance = isRequired(
'contractImplementationInstance'
),
firstBlock = 0,
lastBlock = 'latest'
) =>
Promise.all(
contractInstance
(await contractImplementationInstance.loadContract())
.allEvents({
fromBlock: firstBlock,
toBlock: lastBlock
@@ -51,72 +53,82 @@ class EventListener {

/**
* Fetch all logs from contractInstance for event in range.
* @param {object} contractInstance - truffle-contract instance.
* @param {object} contractImplementationInstance - contract Implementation instance.
* @param {string} eventName - Name of the event.
* @param {number} firstBlock - Lower bound of search range.
* @param {number} lastBlock - Upper bound of search range.
* @returns {Promise} All events in block range.
*/
static getEventLogs = (
contractInstance = isRequired('contractInstance'),
static getEventLogs = async (
contractImplementationInstance = isRequired(
'contractImplementationInstance'
),
eventName = isRequired('eventName'),
firstBlock = 0,
lastBlock = 'latest'
) =>
Promise.all(
contractInstance[eventName]({
fromBlock: firstBlock,
toBlock: lastBlock
}).get((error, result) => {
if (error) throw new Error(errorConstants.ERROR_FETCHING_EVENTS(error))
(await contractImplementationInstance.loadContract())
[eventName]({
fromBlock: firstBlock,
toBlock: lastBlock
})
.get((error, result) => {
if (error)
throw new Error(errorConstants.ERROR_FETCHING_EVENTS(error))

if (eventName === result.event) return result
})
if (eventName === result.event) return result
})
)

/**
* Add contract instance to poll for events.
* @param {object} contractInstance - truffle-contract instance.
* @param {object} contractImplementationInstance - Contract Implementation instance
*/
addContractInstance = contractInstance => {
this.contractInstances.push(contractInstance)
this.contractEventHandlerMap[contractInstance.address] = {}
addContractImplementation = contractImplementationInstance => {
this.contractInstances.push(contractImplementationInstance)
this.contractEventHandlerMap[
contractImplementationInstance.getContractAddress()
] = {}
}

/**
* Remove contract instance. Will also remove all handlers.
* @param {string} contractAddress - Address of contract.
* @param {string} contractImplementationInstance - contract implementation instance
*/
removeContractInstance = (
contractAddress = isRequired('contractAddress')
contractImplementationInstance = isRequired(
'contractImplementationInstance'
)
) => {
const contractAddress = contractImplementationInstance.getContractAddress()
// remove instance from this.contractInstances
const removedInstance = _.remove(
this.contractInstances,
instance => instance.address === contractAddress
instance => instance.getContractAddress() === contractAddress
)
// if we didn't remove anything throw error
if (removedInstance.length === 0)
throw new Error(errorConstants.MISSING_CONTRACT_INSTANCE(contractAddress))
// stop watching on these instances
removedInstance.forEach(instance =>
this.stopWatchingForEvents(instance.address)
)
removedInstance.forEach(instance => this.stopWatchingForEvents(instance))

// remove handlers for contract instance
delete this.contractEventHandlerMap[contractAddress]
}

/**
* Add event handler that will be called when event is broadcasted.
* @param {string} contractAddress - Address of contract.
* @param {string} contractImplementationInstance - Contract implementation instance
* @param {string} eventName - Name of event.
* @param {function} handler - Function to be called when event is consumed.
*/
addEventHandler = (
contractAddress = isRequired('contractAddress'),
contractImplementationInstance = isRequired('contractAddress'),
eventName = isRequired('eventName'),
handler = isRequired('handler')
) => {
const contractAddress = contractImplementationInstance.getContractAddress()
if (!this.contractEventHandlerMap[contractAddress][eventName])
this.contractEventHandlerMap[contractAddress][eventName] = []
this.contractEventHandlerMap[contractAddress][eventName].push(handler)
@@ -125,44 +137,49 @@ class EventListener {
/**
* Watch for events on each contract instance. Call registered handlers on logs
* @param {number} fromBlock - A block number can be passed to catch up on missed logs
* @returns {Promise} - Promise resolves when all watchers have been started
*/
watchForEvents = (fromBlock = 'latest') => {
this.contractInstances.forEach(instance => {
const newWatcherInstance = instance.allEvents({
fromBlock: fromBlock,
lastBlock: 'latest'
})
watchForEvents = async (fromBlock = 'latest') =>
Promise.all(
this.contractInstances.map(async contractImplementation => {
const instance = await contractImplementation.loadContract()
const newWatcherInstance = instance.allEvents({
fromBlock: fromBlock,
lastBlock: 'latest'
})

// NOTE: should we allow more than one listener per contract instance?
if (this.watcherInstances[instance.address])
this.watcherInstances[instance.address].stopWatching()

this.watcherInstances[instance.address] = newWatcherInstance

newWatcherInstance.watch((error, result) => {
if (!error) {
const handlers = this.contractEventHandlerMap[instance.address][
result.event
]
if (handlers) {
handlers.forEach(handler => {
this._queueEvent(handler, result)
})
// NOTE: should we allow more than one listener per contract instance?
if (this.watcherInstances[instance.address])
this.watcherInstances[instance.address].stopWatching()

this.watcherInstances[instance.address] = newWatcherInstance
newWatcherInstance.watch((error, result) => {
if (!error) {
const handlers = this.contractEventHandlerMap[instance.address][
result.event
]
if (handlers) {
handlers.forEach(handler => {
this._queueEvent(handler, result)
})
}
}
}
})
})
})
}
)

/**
* Stop listening on contract. If no contractAddress supplied it stops all listeners
* @param {string} contractAddress - Address of the contract to stop watching
* @param {string} contractImplementationInstance - Address of the contract to stop watching
*/
stopWatchingForEvents = contractAddress => {
if (contractAddress) this.watcherInstances[contractAddress].stopWatching()
stopWatchingForEvents = contractImplementationInstance => {
if (contractImplementationInstance)
this.watcherInstances[
contractImplementationInstance.getContractAddress()
].stopWatching()
else
this.contractInstances.forEach(instance => {
this.watcherInstances[instance.address].stopWatching()
this.watcherInstances[instance.getContractAddress()].stopWatching()
})
}

@@ -69,12 +69,8 @@ describe('Event Listener', () => {
expect(pnkAddress).toBeDefined()

const KlerosPOCInstance = new KlerosPOC(provider, klerosPOCAddress)
// load contract
await KlerosPOCInstance.loadContract()

const EventListenerInstance = new EventListener([
KlerosPOCInstance.getContractInstance()
])
const EventListenerInstance = new EventListener([KlerosPOCInstance])
// set up callback
let { promise: waitPromise, callback: waitCallback } = waitNotifications(
1,
@@ -83,21 +79,19 @@ describe('Event Listener', () => {
// add event handler
const eventName = 'NewPeriod'
EventListenerInstance.addEventHandler(
KlerosPOCInstance.getContractAddress(),
KlerosPOCInstance,
eventName,
waitCallback
)
EventListenerInstance.watchForEvents()
await EventListenerInstance.watchForEvents()

await delaySecond()
KlerosPOCInstance.passPeriod()
// we will wait for 2 seconds for promise to resolve or else throw
let throwError = true
setTimeout(() => {
if (throwError) {
EventListenerInstance.stopWatchingForEvents(
KlerosPOCInstance.getContractAddress()
)
EventListenerInstance.stopWatchingForEvents(KlerosPOCInstance)
throw new Error('Callback Promise did not resolve')
}
}, 1000 * 2)
@@ -109,9 +103,7 @@ describe('Event Listener', () => {
const log = eventLogs[0]
expect(log.event).toEqual(eventName)

EventListenerInstance.stopWatchingForEvents(
KlerosPOCInstance.getContractAddress()
)
EventListenerInstance.stopWatchingForEvents(KlerosPOCInstance)
},
50000
)
@@ -129,30 +121,22 @@ describe('Event Listener', () => {
expect(pnkAddress).toBeDefined()

const KlerosPOCInstance = new KlerosPOC(provider, klerosPOCAddress)
// load contract
await KlerosPOCInstance.loadContract()

// add conract instance with event handler
const EventListenerInstance = new EventListener([
KlerosPOCInstance.getContractInstance()
])
const EventListenerInstance = new EventListener([KlerosPOCInstance])
EventListenerInstance.addEventHandler(
KlerosPOCInstance.getContractAddress(),
KlerosPOCInstance,
'FakeEvent',
() => {}
)

EventListenerInstance.watchForEvents()
await EventListenerInstance.watchForEvents()

EventListenerInstance.removeContractInstance(
KlerosPOCInstance.getContractAddress()
)
EventListenerInstance.removeContractInstance(KlerosPOCInstance)

expect(EventListenerInstance.contractInstances.length).toEqual(0)
expect(
EventListenerInstance.contractEventHandlerMap[
KlerosPOCInstance.getContractAddress()
]
EventListenerInstance.contractEventHandlerMap[KlerosPOCInstance]
).toBeUndefined()
})
})

0 comments on commit 4a89382

Please sign in to comment.
You can’t perform that action at this time.