1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
import * as _ from 'lodash';
import {schemas} from '0x-json-schemas';
import {ZeroEx} from '../';
import {EventWatcher} from './event_watcher';
import {assert} from '../utils/assert';
import {utils} from '../utils/utils';
import {artifacts} from '../artifacts';
import {AbiDecoder} from '../utils/abi_decoder';
import {OrderStateUtils} from '../utils/order_state_utils';
import {
LogEvent,
OrderState,
SignedOrder,
Web3Provider,
BlockParamLiteral,
LogWithDecodedArgs,
OnOrderStateChangeCallback,
ExchangeEvents,
TokenEvents,
ZeroExError,
} from '../types';
import {Web3Wrapper} from '../web3_wrapper';
interface DependentOrderHashes {
[makerAddress: string]: {
[makerToken: string]: Set<string>,
};
}
interface OrderByOrderHash {
[orderHash: string]: SignedOrder;
}
export class OrderStateWatcher {
private _orders: OrderByOrderHash;
private _dependentOrderHashes: DependentOrderHashes;
private _web3Wrapper: Web3Wrapper;
private _callbackAsync?: OnOrderStateChangeCallback;
private _eventWatcher: EventWatcher;
private _abiDecoder: AbiDecoder;
private _orderStateUtils: OrderStateUtils;
constructor(
web3Wrapper: Web3Wrapper, abiDecoder: AbiDecoder, orderStateUtils: OrderStateUtils,
eventPollingIntervalMs?: number) {
this._web3Wrapper = web3Wrapper;
this._orders = {};
this._dependentOrderHashes = {};
this._eventWatcher = new EventWatcher(
this._web3Wrapper, eventPollingIntervalMs,
);
this._abiDecoder = abiDecoder;
this._orderStateUtils = orderStateUtils;
}
/**
* Add an order to the orderStateWatcher
* @param signedOrder The order you wish to start watching.
*/
public addOrder(signedOrder: SignedOrder): void {
assert.doesConformToSchema('signedOrder', signedOrder, schemas.signedOrderSchema);
const orderHash = ZeroEx.getOrderHashHex(signedOrder);
this._orders[orderHash] = signedOrder;
this.addToDependentOrderHashes(signedOrder, orderHash);
}
/**
* Removes an order from the orderStateWatcher
* @param orderHash The orderHash of the order you wish to stop watching.
*/
public removeOrder(orderHash: string): void {
assert.doesConformToSchema('orderHash', orderHash, schemas.orderHashSchema);
const signedOrder = this._orders[orderHash];
if (_.isUndefined(signedOrder)) {
return;
}
delete this._orders[orderHash];
this._dependentOrderHashes[signedOrder.maker][signedOrder.makerTokenAddress].delete(orderHash);
// We currently do not remove the maker/makerToken keys from the mapping when all orderHashes removed
}
/**
* Starts an orderStateWatcher subscription. The callback will be notified every time a watched order's
* backing blockchain state has changed. This is a call-to-action for the caller to re-validate the order
* @param callback Receives the orderHash of the order that should be re-validated, together.
* with all the order-relevant blockchain state needed to re-validate the order
* @param numConfirmations Number of confirmed blocks deeps you want to run the orderWatcher from. Passing
* is 0 will watch the backing node's mempool, 3 will emit events when blockchain
* state relevant to a watched order changed 3 blocks ago.
*/
public subscribe(callback: OnOrderStateChangeCallback, numConfirmations: number): void {
assert.isFunction('callback', callback);
if (!_.isUndefined(this._callbackAsync)) {
throw new Error(ZeroExError.SubscriptionAlreadyPresent);
}
this._callbackAsync = callback;
this._eventWatcher.subscribe(this._onEventWatcherCallbackAsync.bind(this), numConfirmations);
}
/**
* Ends an orderStateWatcher subscription.
* @param signedOrder The order you wish to stop watching.
*/
public unsubscribe(): void {
delete this._callbackAsync;
this._eventWatcher.unsubscribe();
}
private async _onEventWatcherCallbackAsync(log: LogEvent): Promise<void> {
const maybeDecodedLog = this._abiDecoder.tryToDecodeLogOrNoop(log);
const isDecodedLog = !_.isUndefined((maybeDecodedLog as LogWithDecodedArgs<any>).event);
if (!isDecodedLog) {
return; // noop
}
const decodedLog = maybeDecodedLog as LogWithDecodedArgs<any>;
let makerToken: string;
let makerAddress: string;
let orderHashesSet: Set<string>;
switch (decodedLog.event) {
case TokenEvents.Approval:
makerToken = decodedLog.address;
makerAddress = decodedLog.args._owner;
orderHashesSet = _.get(this._dependentOrderHashes, [makerAddress, makerToken]);
if (!_.isUndefined(orderHashesSet)) {
const orderHashes = Array.from(orderHashesSet);
await this._emitRevalidateOrdersAsync(orderHashes);
}
break;
case TokenEvents.Transfer:
makerToken = decodedLog.address;
makerAddress = decodedLog.args._from;
orderHashesSet = _.get(this._dependentOrderHashes, [makerAddress, makerToken]);
if (!_.isUndefined(orderHashesSet)) {
const orderHashes = Array.from(orderHashesSet);
await this._emitRevalidateOrdersAsync(orderHashes);
}
break;
case ExchangeEvents.LogFill:
case ExchangeEvents.LogCancel:
const orderHash = decodedLog.args.orderHash;
const isOrderWatched = !_.isUndefined(this._orders[orderHash]);
if (isOrderWatched) {
await this._emitRevalidateOrdersAsync([orderHash]);
}
break;
case ExchangeEvents.LogError:
return; // noop
default:
throw utils.spawnSwitchErr('decodedLog.event', decodedLog.event);
}
}
private async _emitRevalidateOrdersAsync(orderHashes: string[]): Promise<void> {
// TODO: Make defaultBlock a passed in option
const methodOpts = {
defaultBlock: BlockParamLiteral.Pending,
};
for (const orderHash of orderHashes) {
const signedOrder = this._orders[orderHash] as SignedOrder;
const orderState = await this._orderStateUtils.getOrderStateAsync(signedOrder, methodOpts);
if (!_.isUndefined(this._callbackAsync)) {
await this._callbackAsync(orderState);
} else {
break; // Unsubscribe was called
}
}
}
private addToDependentOrderHashes(signedOrder: SignedOrder, orderHash: string) {
if (_.isUndefined(this._dependentOrderHashes[signedOrder.maker])) {
this._dependentOrderHashes[signedOrder.maker] = {};
}
if (_.isUndefined(this._dependentOrderHashes[signedOrder.maker][signedOrder.makerTokenAddress])) {
this._dependentOrderHashes[signedOrder.maker][signedOrder.makerTokenAddress] = new Set();
}
this._dependentOrderHashes[signedOrder.maker][signedOrder.makerTokenAddress].add(orderHash);
}
}
|
