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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
import * as _ from 'lodash';
import {schemas} from '0x-json-schemas';
import * as ethUtil from 'ethereumjs-util';
import {ZeroEx} from '../0x';
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,
OrderStateWatcherConfig,
ExchangeEvents,
TokenEvents,
ZeroExError,
} from '../types';
import {Web3Wrapper} from '../web3_wrapper';
const DEFAULT_NUM_CONFIRMATIONS = 0;
interface DependentOrderHashes {
[makerAddress: string]: {
[makerToken: string]: Set<string>,
};
}
interface OrderByOrderHash {
[orderHash: string]: SignedOrder;
}
/**
* This class includes all the functionality related to watching a set of orders
* for potential changes in order validity/fillability. The orderWatcher notifies
* the subscriber of these changes so that a final decison can be made on whether
* the order should be deemed invalid.
*/
export class OrderStateWatcher {
private _orderByOrderHash: OrderByOrderHash = {};
private _dependentOrderHashes: DependentOrderHashes = {};
private _web3Wrapper: Web3Wrapper;
private _callbackIfExistsAsync?: OnOrderStateChangeCallback;
private _eventWatcher: EventWatcher;
private _abiDecoder: AbiDecoder;
private _orderStateUtils: OrderStateUtils;
private _numConfirmations: number;
constructor(
web3Wrapper: Web3Wrapper, abiDecoder: AbiDecoder, orderStateUtils: OrderStateUtils,
config?: OrderStateWatcherConfig,
) {
this._web3Wrapper = web3Wrapper;
const eventPollingIntervalMs = _.isUndefined(config) ? undefined : config.pollingIntervalMs;
this._numConfirmations = _.isUndefined(config) ?
DEFAULT_NUM_CONFIRMATIONS
: config.numConfirmations;
this._eventWatcher = new EventWatcher(
this._web3Wrapper, eventPollingIntervalMs, this._numConfirmations,
);
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);
assert.isValidSignature(orderHash, signedOrder.ecSignature, signedOrder.maker);
this._orderByOrderHash[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._orderByOrderHash[orderHash];
if (_.isUndefined(signedOrder)) {
return; // noop
}
delete this._orderByOrderHash[orderHash];
this.removeFromDependentOrderHashes(signedOrder.maker, signedOrder.makerTokenAddress, orderHash);
}
/**
* Starts an orderStateWatcher subscription. The callback will be called 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.
*/
public subscribe(callback: OnOrderStateChangeCallback): void {
assert.isFunction('callback', callback);
if (!_.isUndefined(this._callbackIfExistsAsync)) {
throw new Error(ZeroExError.SubscriptionAlreadyPresent);
}
this._callbackIfExistsAsync = callback;
this._eventWatcher.subscribe(this._onEventWatcherCallbackAsync.bind(this));
}
/**
* Ends an orderStateWatcher subscription.
*/
public unsubscribe(): void {
delete this._callbackIfExistsAsync;
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 blockNumberBuff = ethUtil.toBuffer(maybeDecodedLog.blockNumber);
const blockNumber = ethUtil.bufferToInt(blockNumberBuff);
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, blockNumber);
}
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, blockNumber);
}
break;
case ExchangeEvents.LogFill:
case ExchangeEvents.LogCancel:
const orderHash = decodedLog.args.orderHash;
const isOrderWatched = !_.isUndefined(this._orderByOrderHash[orderHash]);
if (isOrderWatched) {
await this._emitRevalidateOrdersAsync([orderHash], blockNumber);
}
break;
case ExchangeEvents.LogError:
return; // noop
default:
throw utils.spawnSwitchErr('decodedLog.event', decodedLog.event);
}
}
private async _emitRevalidateOrdersAsync(orderHashes: string[], blockNumber: number): Promise<void> {
const defaultBlock = this._numConfirmations === 0 ?
BlockParamLiteral.Pending :
blockNumber;
const methodOpts = {
defaultBlock,
};
for (const orderHash of orderHashes) {
const signedOrder = this._orderByOrderHash[orderHash] as SignedOrder;
const orderState = await this._orderStateUtils.getOrderStateAsync(signedOrder, methodOpts);
if (!_.isUndefined(this._callbackIfExistsAsync)) {
await this._callbackIfExistsAsync(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);
}
private removeFromDependentOrderHashes(makerAddress: string, makerTokenAddress: string, orderHash: string) {
this._dependentOrderHashes[makerAddress][makerTokenAddress].delete(orderHash);
if (this._dependentOrderHashes[makerAddress][makerTokenAddress].size === 0) {
delete this._dependentOrderHashes[makerAddress][makerTokenAddress];
}
if (_.isEmpty(this._dependentOrderHashes[makerAddress])) {
delete this._dependentOrderHashes[makerAddress];
}
}
}
|
