Skip to content

Commit 4c119d5

Browse files
committed
refactor: converting NodeConnection
* Related #512 * Related #495 * Related #234 [ci skip]
1 parent 045be37 commit 4c119d5

10 files changed

+113
-168
lines changed

src/nodes/NodeConnection.ts

Lines changed: 65 additions & 110 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,27 @@
11
import type { ContextTimed } from '@matrixai/contexts';
22
import type { PromiseCancellable } from '@matrixai/async-cancellable';
33
import type { NodeId } from './types';
4-
import type Proxy from '../network/Proxy';
54
import type { Host, Hostname, Port } from '../network/types';
65
import type { Certificate } from '../keys/types';
7-
import type GRPCClient from '../grpc/GRPCClient';
6+
import type { ClientManifest } from '@/rpc/types';
7+
import type { Host as QUICHost, Port as QUICPort } from '@matrixai/quic';
8+
import type { QUICClientConfig } from './types';
89
import Logger from '@matrixai/logger';
910
import { CreateDestroy, ready } from '@matrixai/async-init/dist/CreateDestroy';
1011
import * as asyncInit from '@matrixai/async-init';
1112
import { timedCancellable, context } from '@matrixai/contexts/dist/decorators';
12-
import * as nodesErrors from './errors';
13-
import * as grpcErrors from '../grpc/errors';
13+
import { QUICClient } from '@matrixai/quic';
14+
import RPCClient from '@/rpc/RPCClient';import * as nodesErrors from './errors';
1415
import * as networkUtils from '../network/utils';
15-
import { timerStart } from '../utils/index';
16+
import * as rpcUtils from '../rpc/utils';
1617

18+
// TODO: extend an event system, use events for cleaning up.
1719
/**
1820
* Encapsulates the unidirectional client-side connection of one node to another.
1921
*/
20-
// eslint-disable-next-line @typescript-eslint/no-unused-vars -- False positive for T
21-
interface NodeConnection<T extends GRPCClient> extends CreateDestroy {}
22+
interface NodeConnection<M extends ClientManifest> extends CreateDestroy {}
2223
@CreateDestroy()
23-
class NodeConnection<T extends GRPCClient> {
24+
class NodeConnection<M extends ClientManifest> {
2425
public readonly host: Host;
2526
public readonly port: Port;
2627
/**
@@ -30,124 +31,84 @@ class NodeConnection<T extends GRPCClient> {
3031
public readonly hostname?: Hostname;
3132

3233
protected logger: Logger;
33-
protected destroyCallback: () => Promise<void>;
34-
protected proxy: Proxy;
35-
protected client: T;
34+
protected quicClient: QUICClient;
35+
protected rpcClient: RPCClient<M>;
3636

37-
static createNodeConnection<T extends GRPCClient>(
37+
static createNodeConnection<M extends ClientManifest>(
3838
{
3939
targetNodeId,
4040
targetHost,
4141
targetPort,
4242
targetHostname,
43-
proxy,
44-
clientFactory,
45-
destroyCallback,
46-
destroyTimeout,
43+
quicClientConfig,
44+
manifest,
4745
logger,
4846
}: {
4947
targetNodeId: NodeId;
5048
targetHost: Host;
5149
targetPort: Port;
5250
targetHostname?: Hostname;
53-
proxy: Proxy;
54-
clientFactory: (...args) => Promise<T>;
55-
destroyCallback?: () => Promise<void>;
56-
destroyTimeout?: number;
51+
quicClientConfig: QUICClientConfig;
52+
manifest: M;
5753
logger?: Logger;
5854
},
5955
ctx?: Partial<ContextTimed>,
60-
): PromiseCancellable<NodeConnection<T>>;
56+
): PromiseCancellable<NodeConnection<M>>;
6157
@timedCancellable(true, 20000)
62-
static async createNodeConnection<T extends GRPCClient>(
58+
static async createNodeConnection<M extends ClientManifest>(
6359
{
6460
targetNodeId,
6561
targetHost,
6662
targetPort,
6763
targetHostname,
68-
proxy,
69-
clientFactory,
70-
destroyCallback = async () => {},
71-
destroyTimeout = 2000,
64+
quicClientConfig,
65+
manifest,
7266
logger = new Logger(this.name),
7367
}: {
7468
targetNodeId: NodeId;
7569
targetHost: Host;
7670
targetPort: Port;
7771
targetHostname?: Hostname;
78-
proxy: Proxy;
79-
clientFactory: (...args) => Promise<T>;
80-
destroyCallback?: () => Promise<void>;
81-
destroyTimeout?: number;
72+
quicClientConfig: QUICClientConfig;
73+
manifest: M;
8274
logger?: Logger;
8375
},
8476
@context ctx: ContextTimed,
85-
): Promise<NodeConnection<T>> {
77+
): Promise<NodeConnection<M>> {
8678
logger.info(`Creating ${this.name}`);
8779
// Checking if attempting to connect to a wildcard IP
8880
if (networkUtils.isHostWildcard(targetHost)) {
8981
throw new nodesErrors.ErrorNodeConnectionHostWildcard();
9082
}
91-
const proxyConfig = {
92-
host: proxy.getForwardHost(),
93-
port: proxy.getForwardPort(),
94-
authToken: proxy.authToken,
95-
};
96-
// 1. Ask fwdProxy for connection to target (the revProxy of other node)
97-
// 2. Start sending hole-punching packets to the target (via the client start -
98-
// this establishes a HTTP CONNECT request with the forward proxy)
99-
// 3. Relay the proxy port to the broker/s (such that they can inform the other node)
100-
// 4. Start sending hole-punching packets to other node (done in openConnection())
101-
// Done in parallel
102-
const nodeConnection = new this<T>({
83+
// TODO: this needs to be updated to take a context,
84+
// still uses old timer style.
85+
const clientLogger = logger.getChild(RPCClient.name);
86+
// TODO: Custom TLS validation with NodeId
87+
// TODO: Idle timeout and connection timeout is the same thing from the `quic` perspective.
88+
// THis means we need to race our timeout timer
89+
const quicClient = await QUICClient.createQUICClient({
90+
host: targetHost as unknown as QUICHost, // FIXME: better type conversion?
91+
port: targetPort as unknown as QUICPort, // FIXME: better type conversion?
92+
...quicClientConfig,
93+
logger: logger.getChild(QUICClient.name),
94+
});
95+
const rpcClient = await RPCClient.createRPCClient<M>({
96+
manifest,
97+
middlewareFactory: rpcUtils.defaultClientMiddlewareWrapper(),
98+
streamFactory: () => {
99+
return quicClient.connection.streamNew();
100+
},
101+
logger: clientLogger,
102+
});
103+
const nodeConnection = new this<M>({
103104
host: targetHost,
104105
port: targetPort,
105106
hostname: targetHostname,
106-
proxy: proxy,
107-
destroyCallback,
107+
quicClient,
108+
rpcClient,
108109
logger,
109110
});
110-
let client: T;
111-
try {
112-
// TODO: this needs to be updated to take a context,
113-
// still uses old timer style.
114-
const clientLogger = logger.getChild(clientFactory.name);
115-
client = await clientFactory({
116-
nodeId: targetNodeId,
117-
host: targetHost,
118-
port: targetPort,
119-
proxyConfig: proxyConfig,
120-
// Think about this
121-
logger: clientLogger,
122-
destroyCallback: async () => {
123-
clientLogger.debug(`GRPC client triggered destroyedCallback`);
124-
if (
125-
nodeConnection[asyncInit.status] !== 'destroying' &&
126-
!nodeConnection[asyncInit.destroyed]
127-
) {
128-
await nodeConnection.destroy({ timeout: destroyTimeout });
129-
}
130-
},
131-
// FIXME: this needs to be replaced with
132-
// the GRPC timerCancellable update
133-
timer: timerStart(ctx.timer.getTimeout()),
134-
});
135-
// 5. When finished, you have a connection to other node
136-
// The GRPCClient is ready to be used for requests
137-
} catch (e) {
138-
await nodeConnection.destroy({ timeout: destroyTimeout });
139-
// If the connection times out, re-throw this with a higher level nodes exception
140-
if (e instanceof grpcErrors.ErrorGRPCClientTimeout) {
141-
throw new nodesErrors.ErrorNodeConnectionTimeout(e.message, {
142-
cause: e,
143-
});
144-
}
145-
throw e;
146-
}
147-
// FIXME: we need a finally block here to do cleanup.
148-
// TODO: This is due to chicken or egg problem
149-
// see if we can move to CreateDestroyStartStop to resolve this
150-
nodeConnection.client = client;
111+
nodeConnection.rpcClient = rpcClient;
151112
logger.info(`Created ${this.name}`);
152113
return nodeConnection;
153114
}
@@ -156,48 +117,43 @@ class NodeConnection<T extends GRPCClient> {
156117
host,
157118
port,
158119
hostname,
159-
proxy,
160-
destroyCallback,
120+
quicClient,
121+
rpcClient,
161122
logger,
162123
}: {
163124
host: Host;
164125
port: Port;
165126
hostname?: Hostname;
166-
proxy: Proxy;
167-
destroyCallback: () => Promise<void>;
127+
quicClient: QUICClient;
128+
rpcClient: RPCClient<M>;
168129
logger: Logger;
169130
}) {
170131
this.logger = logger;
171132
this.host = host;
172133
this.port = port;
173134
this.hostname = hostname;
174-
this.proxy = proxy;
175-
this.destroyCallback = destroyCallback;
135+
this.quicClient = quicClient;
136+
this.rpcClient = rpcClient;
176137
}
177138

178139
public async destroy({
179-
timeout,
140+
force,
180141
}: {
181-
timeout?: number;
142+
force?: boolean;
182143
} = {}) {
183144
this.logger.info(`Destroying ${this.constructor.name}`);
184-
if (
185-
this.client != null &&
186-
this.client[asyncInit.status] !== 'destroying' &&
187-
!this.client[asyncInit.destroyed]
188-
) {
189-
await this.client.destroy({ timeout });
190-
}
191-
this.logger.debug(`${this.constructor.name} triggered destroyedCallback`);
192-
await this.destroyCallback();
145+
await this.quicClient.destroy({ force });
146+
await this.rpcClient.destroy();
147+
this.logger.debug(`${this.constructor.name} triggered destroyed event`);
148+
// TODO: trigger destroy event
193149
this.logger.info(`Destroyed ${this.constructor.name}`);
194150
}
195151

196152
/**
197153
* Gets GRPCClient for this node connection
198154
*/
199-
public getClient(): T {
200-
return this.client;
155+
public getClient(): RPCClient<M> {
156+
return this.rpcClient;
201157
}
202158

203159
/**
@@ -207,11 +163,10 @@ class NodeConnection<T extends GRPCClient> {
207163
*/
208164
@ready(new nodesErrors.ErrorNodeConnectionDestroyed())
209165
public getRootCertChain(): Array<Certificate> {
210-
const connInfo = this.proxy.getConnectionInfoByProxy(this.host, this.port);
211-
if (connInfo == null) {
212-
throw new nodesErrors.ErrorNodeConnectionInfoNotExist();
213-
}
214-
return connInfo.remoteCertificates;
166+
const connInfo = this.quicClient.connection.remoteInfo;
167+
// TODO:
168+
throw Error('TMP IMP');
169+
// Return connInfo.remoteCertificates;
215170
}
216171
}
217172

0 commit comments

Comments
 (0)