Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Access list sub #326

Merged
merged 3 commits into from
Oct 22, 2024
Merged

Access list sub #326

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
36d4d2a
wip
robertlincecum Oct 18, 2024
4742062
wip
robertlincecum Oct 21, 2024
1f9ba0a
balance and block accesses subs working
robertlincecum Oct 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 42 additions & 1 deletion dist/quais.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30191,7 +30191,7 @@ async function getSubscription(_event, zone) {
// @todo Should lowercase and whatnot things here instead of copy...
return { type: 'orphan', tag: getTag('orphan', event), filter: copy$1(event), zone };
}
if (_event.address || _event.topics) {
if (_event.topics || Array.isArray(_event.address)) {
const event = _event;
const filter = {
topics: (event.topics || []).map((t) => {
Expand Down Expand Up @@ -30244,6 +30244,16 @@ async function getSubscription(_event, zone) {
}
return { filter, tag: getTag('event', filter), type: 'event', zone };
}
else if (_event.address) {
const address = formatMixedCaseChecksumAddress(isHexString(_event.address) ? _event.address : await resolveAddress(_event.address));
const filter = {
address: address
};
if (!zone) {
zone = toZone(address.slice(0, 4));
}
return { filter, tag: getTag('accesses', filter), type: 'accesses', zone };
}
assertArgument(false, 'unknown ProviderEvent', 'event', _event);
}
/**
Expand Down Expand Up @@ -33521,6 +33531,35 @@ class SocketBlockSubscriber extends SocketSubscriber {
provider.emit('block', this.zone, parseInt(message.woHeader.number));
}
}
/**
* A **SocketAccessesSubscriber** listens for `acceses` events and emits `accesses` events.
*
* @category Providers
*/
class SocketAccessesSubscriber extends SocketSubscriber {
/**
* Creates a new **SocketBlockSubscriber**.
*
* @ignore
* @param {SocketProvider} provider - The socket provider.
* @param filter
* @param zone
*/
constructor(provider, filter, zone) {
super(provider, ['accesses', filter.address], zone);
}
/**
* Emit the block event.
*
* @ignore
* @param {SocketProvider} provider - The socket provider.
* @param {any} message - The message to emit.
* @returns {Promise<void>}
*/
async _emit(provider, message) {
provider.emit('accesses', this.zone, message);
}
}
/**
* A **SocketPendingSubscriber** listens for pending transactions and emits `"pending"` events.
*
Expand Down Expand Up @@ -33639,6 +33678,8 @@ class SocketProvider extends JsonRpcApiProvider {
return new UnmanagedSubscriber('close');
case 'block':
return new SocketBlockSubscriber(this, sub.zone);
case 'accesses':
return new SocketAccessesSubscriber(this, sub.filter, sub.zone);
case 'pending':
return new SocketPendingSubscriber(this, sub.zone);
case 'event':
Expand Down
2 changes: 1 addition & 1 deletion dist/quais.js.map
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion dist/quais.min.js
View file
Open in desktop

Large diffs are not rendered by default.

43 changes: 42 additions & 1 deletion dist/quais.umd.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30213,7 +30213,7 @@ const __$G = (typeof globalThis !== 'undefined' ? globalThis: typeof window !==
// @todo Should lowercase and whatnot things here instead of copy...
return { type: 'orphan', tag: getTag('orphan', event), filter: copy$1(event), zone };
}
if (_event.address || _event.topics) {
if (_event.topics || Array.isArray(_event.address)) {
const event = _event;
const filter = {
topics: (event.topics || []).map((t) => {
Expand Down Expand Up @@ -30266,6 +30266,16 @@ const __$G = (typeof globalThis !== 'undefined' ? globalThis: typeof window !==
}
return { filter, tag: getTag('event', filter), type: 'event', zone };
}
else if (_event.address) {
const address = formatMixedCaseChecksumAddress(isHexString(_event.address) ? _event.address : await resolveAddress(_event.address));
const filter = {
address: address
};
if (!zone) {
zone = toZone(address.slice(0, 4));
}
return { filter, tag: getTag('accesses', filter), type: 'accesses', zone };
}
assertArgument(false, 'unknown ProviderEvent', 'event', _event);
}
/**
Expand Down Expand Up @@ -33543,6 +33553,35 @@ const __$G = (typeof globalThis !== 'undefined' ? globalThis: typeof window !==
provider.emit('block', this.zone, parseInt(message.woHeader.number));
}
}
/**
* A **SocketAccessesSubscriber** listens for `acceses` events and emits `accesses` events.
*
* @category Providers
*/
class SocketAccessesSubscriber extends SocketSubscriber {
/**
* Creates a new **SocketBlockSubscriber**.
*
* @ignore
* @param {SocketProvider} provider - The socket provider.
* @param filter
* @param zone
*/
constructor(provider, filter, zone) {
super(provider, ['accesses', filter.address], zone);
}
/**
* Emit the block event.
*
* @ignore
* @param {SocketProvider} provider - The socket provider.
* @param {any} message - The message to emit.
* @returns {Promise<void>}
*/
async _emit(provider, message) {
provider.emit('accesses', this.zone, message);
}
}
/**
* A **SocketPendingSubscriber** listens for pending transactions and emits `"pending"` events.
*
Expand Down Expand Up @@ -33661,6 +33700,8 @@ const __$G = (typeof globalThis !== 'undefined' ? globalThis: typeof window !==
return new UnmanagedSubscriber('close');
case 'block':
return new SocketBlockSubscriber(this, sub.zone);
case 'accesses':
return new SocketAccessesSubscriber(this, sub.filter, sub.zone);
case 'pending':
return new SocketPendingSubscriber(this, sub.zone);
case 'event':
Expand Down
2 changes: 1 addition & 1 deletion dist/quais.umd.js.map
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion dist/quais.umd.min.js
View file
Open in desktop

Large diffs are not rendered by default.

32 changes: 19 additions & 13 deletions src/_tests/integration/sendquai.integration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import assert from 'assert';

import { JsonRpcProvider, Wallet } from '../../index.js';
import { WebSocketProvider, Wallet } from '../../index.js';

import dotenv from 'dotenv';
import { QuaiTransactionResponse } from '../../providers/provider.js';
Expand All @@ -17,27 +17,28 @@ const testCases = [
receiverAddressEnvVar: 'CYPRUS1_ADDR_2',
skipReceiverBalanceCheck: false,
},
{
description: 'Cyprus1 to Cyprus2',
receiverAddressEnvVar: 'CYPRUS2_ADDR_1',
skipReceiverBalanceCheck: true,
},
{
description: 'Cyprus1 to Paxos1',
receiverAddressEnvVar: 'PAXOS1_ADDR_1',
skipReceiverBalanceCheck: true,
},
// {
// description: 'Cyprus1 to Cyprus2',
// receiverAddressEnvVar: 'CYPRUS2_ADDR_1',
// skipReceiverBalanceCheck: true,
// },
// {
// description: 'Cyprus1 to Paxos1',
// receiverAddressEnvVar: 'PAXOS1_ADDR_1',
// skipReceiverBalanceCheck: true,
// },
];

describe('Test sending Quai', function () {
this.timeout(120000);

let provider: JsonRpcProvider;
let provider: WebSocketProvider;
let wallet: Wallet;
const quaiAmount = 42000000000n;

before(async () => {
provider = new JsonRpcProvider(process.env.RPC_URL);
const wsUrl = process.env.RPC_URL?.replace('https', 'wss')?.replace('http', 'ws');
provider = new WebSocketProvider(wsUrl ?? '');
wallet = new Wallet(process.env.CYPRUS1_PRIVKEY_1!, provider);
const senderBalance = await provider.getBalance(wallet.address);
// ensure balance is greater than 0.1 QUAI
Expand All @@ -63,7 +64,12 @@ describe('Test sending Quai', function () {
from: wallet.address,
};
console.log(`Sending quai to: ${receiverAddress}`);
provider.on({ type: 'balance', address: receiverAddress }, (balance) => {
console.log(`Received quai on address ${receiverAddress}. New Balance is ${balance}`);
});
const tx = (await wallet.sendTransaction(txObj)) as QuaiTransactionResponse;
//wait 2 seconds
await new Promise((resolve) => setTimeout(resolve, 10000));
assert(tx);
console.log('Waiting for Quai Tx to be mined...');
const receipt = await tx.wait();
Expand Down
21 changes: 20 additions & 1 deletion src/providers/abstract-provider.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ import {
QuaiTransactionResponse,
QiTransactionResponse,
QuaiTransactionRequest,
AccessesFilter,
} from './provider.js';

import type { Addressable, AddressLike } from '../address/index.js';
Expand Down Expand Up @@ -166,6 +167,12 @@ export type Subscription =
hash: string;
zone: Zone;
}
| {
type: 'accesses';
tag: string;
filter: AccessesFilter;
zone: Zone;
}
| {
type: 'qiTransaction';
tag: string;
Expand Down Expand Up @@ -355,7 +362,19 @@ async function getSubscription(_event: ProviderEvent, zone?: Zone): Promise<Subs
return { type: 'orphan', tag: getTag('orphan', event), filter: copy(event), zone };
}

if ((<any>_event).address || (<any>_event).topics) {
if ((<any>_event).type && (<any>_event).address) {
const address = formatMixedCaseChecksumAddress(
isHexString((<any>_event).address) ? (<any>_event).address : await resolveAddress((<any>_event).address),
);
const filter = <AccessesFilter>{
type: (<any>_event).type,
address: address,
};
if (!zone) {
zone = toZone(address.slice(0, 4));
}
return { filter, tag: getTag('accesses', filter), type: 'accesses', zone };
} else if ((<any>_event).topics || (<any>_event).address) {
const event = <EventFilter>_event;

const filter: EventFilter = {
Expand Down
44 changes: 43 additions & 1 deletion src/providers/provider-socket.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ import { assert, assertArgument, makeError } from '../utils/index.js';
import { JsonRpcApiProvider } from './provider-jsonrpc.js';

import type { Subscriber, Subscription } from './abstract-provider.js';
import type { EventFilter } from './provider.js';
import type { AccessesFilter, EventFilter } from './provider.js';
import type { JsonRpcApiProviderOptions, JsonRpcError, JsonRpcPayload, JsonRpcResult } from './provider-jsonrpc.js';
import type { Networkish } from './network.js';
import type { WebSocketLike } from './provider-websocket.js';
Expand Down Expand Up @@ -182,6 +182,46 @@ export class SocketBlockSubscriber extends SocketSubscriber {
}
}

/**
* A **SocketAccessesSubscriber** listens for `acceses` events and emits `accesses` events.
*
* @category Providers
*/
export class SocketAccessesSubscriber extends SocketSubscriber {
#accessesFilter: string;

get accessesFilter(): AccessesFilter {
return JSON.parse(this.#accessesFilter);
}
/**
* Creates a new **SocketBlockSubscriber**.
*
* @ignore
* @param {SocketProvider} provider - The socket provider.
* @param filter
* @param zone
*/
constructor(provider: SocketProvider, filter: AccessesFilter, zone: Zone) {
super(provider, ['accesses', filter.address], zone);
this.#accessesFilter = JSON.stringify(filter);
}

/**
* Emit the block event.
*
* @ignore
* @param {SocketProvider} provider - The socket provider.
* @param {any} message - The message to emit.
* @returns {Promise<void>}
*/
async _emit(provider: SocketProvider, message: any): Promise<void> {
if (this.accessesFilter.type === 'balance') {
message = await provider.getBalance(this.accessesFilter.address);
}
provider.emit(this.accessesFilter, this.zone, message);
}
}

/**
* A **SocketPendingSubscriber** listens for pending transactions and emits `"pending"` events.
*
Expand Down Expand Up @@ -318,6 +358,8 @@ export class SocketProvider extends JsonRpcApiProvider<WebSocketLike> {
return new UnmanagedSubscriber('close');
case 'block':
return new SocketBlockSubscriber(this, sub.zone);
case 'accesses':
return new SocketAccessesSubscriber(this, sub.filter, sub.zone);
case 'pending':
return new SocketPendingSubscriber(this, sub.zone);
case 'event':
Expand Down
30 changes: 23 additions & 7 deletions src/providers/provider-websocket.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ export interface WebSocketLike {
onopen: null | ((...args: Array<any>) => any);
onmessage: null | ((...args: Array<any>) => any);
onerror: null | ((...args: Array<any>) => any);
onclose: null | ((...args: Array<any>) => any);

readyState: number;

Expand Down Expand Up @@ -116,6 +117,20 @@ export class WebSocketProvider extends SocketProvider {
// @TODO: now what? Attempt reconnect?
}
};

// @TODO: implement onclose
// websocket.onclose = () => {
// console.log('WebSocket closed. Attempting to reconnect...');
// setTimeout(() => {
// const baseUrl = websocket.url.split(':').slice(0, 2).join(':');
// const shardSuffix = this._getOption('usePathing') ? `/${fromShard(shard, 'nickname')}` : `:${port}`;
// const newWebSocket = this.createWebSocket(baseUrl, shardSuffix);
// this.initWebSocket(newWebSocket, shard, port);
// this.#websockets.push(newWebSocket);
// this._urlMap.set(shard, newWebSocket);
// }, 500); // Reconnect after 5 seconds
// };

websocket.onmessage = (message: { data: string }) => {
this._processMessage(message.data);
};
Expand All @@ -139,6 +154,12 @@ export class WebSocketProvider extends SocketProvider {
}
}

createWebSocket = (baseUrl: string, suffix: string): WebSocketLike => {
const tempWs = new _WebSocket(`${baseUrl}${suffix}`);
return tempWs as WebSocketLike;
// wait 2 minutes
};

/**
* Initialize the URL map with WebSocket connections.
*
Expand All @@ -152,11 +173,6 @@ export class WebSocketProvider extends SocketProvider {
this._urlMap.clear();
try {
const primeSuffix = this._getOption('usePathing') ? `/${fromShard(Shard.Prime, 'nickname')}` : ':8001';
const createWebSocket = (baseUrl: string, suffix: string): WebSocketLike => {
const tempWs = new _WebSocket(`${baseUrl}${suffix}`);
return tempWs as WebSocketLike;
// wait 2 minutes
};

const initShardWebSockets = async (baseUrl: string) => {
const shards = await this._getRunningLocations(Shard.Prime, true);
Expand All @@ -168,7 +184,7 @@ export class WebSocketProvider extends SocketProvider {
? `/${fromShard(shardEnum, 'nickname')}`
: `:${port}`;
const shardUrl = baseUrl.split(':').slice(0, 2).join(':');
const websocket = createWebSocket(shardUrl, shardSuffix);
const websocket = this.createWebSocket(shardUrl, shardSuffix);
this.initWebSocket(websocket, shardEnum);
this.#websockets.push(websocket);
this._urlMap.set(shardEnum, websocket);
Expand All @@ -185,7 +201,7 @@ export class WebSocketProvider extends SocketProvider {
if (Array.isArray(urls)) {
for (const url of urls) {
const baseUrl = `${url.split(':')[0]}:${url.split(':')[1]}`;
const primeWebsocket = createWebSocket(baseUrl, primeSuffix);
const primeWebsocket = this.createWebSocket(baseUrl, primeSuffix);
this.initWebSocket(primeWebsocket, Shard.Prime);
this.#websockets.push(primeWebsocket);
this._urlMap.set(Shard.Prime, primeWebsocket);
Expand Down
Loading
Loading