This commit is contained in:
Nathan Bierema 2022-06-12 20:43:48 -04:00
parent 88b1b1017a
commit ebf3625311
3 changed files with 177 additions and 141 deletions

View File

@ -1,4 +1,4 @@
import socketCluster, { SCClientSocket } from 'socketcluster-client';
import socketCluster, { AGClientSocket } from 'socketcluster-client';
import { stringify } from 'jsan';
import { Dispatch, MiddlewareAPI } from 'redux';
import * as actions from '../constants/socketActionTypes';
@ -25,11 +25,16 @@ import {
import { nonReduxDispatch } from '../utils/monitorActions';
import { StoreState } from '../reducers';
let socket: SCClientSocket;
let socket: AGClientSocket;
let store: MiddlewareAPI<Dispatch<StoreAction>, StoreState>;
function emit({ message: type, id, instanceId, action, state }: EmitAction) {
socket.emit(id ? `sc-${id}` : 'respond', { type, action, state, instanceId });
socket.transmit(id ? `sc-${id}` : 'respond', {
type,
action,
state,
instanceId,
});
}
function startMonitoring(channel: string) {
@ -120,7 +125,7 @@ function monitoring(request: MonitoringRequest) {
instanceId === instances.selected &&
(request.type === 'ACTION' || request.type === 'STATE')
) {
socket.emit('respond', {
socket.transmit('respond', {
type: 'SYNC',
state: stringify(instances.states[instanceId]),
id: request.id,
@ -134,58 +139,78 @@ function subscribe(
subscription: typeof UPDATE_STATE | typeof UPDATE_REPORTS
) {
const channel = socket.subscribe(channelName);
if (subscription === UPDATE_STATE) channel.watch(monitoring);
else {
if (subscription === UPDATE_STATE) {
void (async () => {
for await (const data of channel) {
monitoring(data as MonitoringRequest);
}
})();
} else {
const watcher = (request: UpdateReportsRequest) => {
store.dispatch({ type: subscription, request });
};
channel.watch(watcher);
socket.on(channelName, watcher);
void (async () => {
for await (const data of channel) {
watcher(data as UpdateReportsRequest);
}
})();
}
}
function handleConnection() {
socket.on('connect', (status) => {
store.dispatch({
type: actions.CONNECT_SUCCESS,
payload: {
id: status.id,
authState: socket.authState,
socketState: socket.state,
},
error: status.authError,
});
if (socket.authState !== actions.AUTHENTICATED) {
store.dispatch({ type: actions.AUTH_REQUEST });
void (async () => {
for await (const data of socket.listener('connect')) {
store.dispatch({
type: actions.CONNECT_SUCCESS,
payload: {
id: data.id,
authState: socket.authState,
socketState: socket.state,
},
// TODO Fix
// @ts-expect-error Because
error: data.authError,
});
if (socket.authState !== actions.AUTHENTICATED) {
store.dispatch({ type: actions.AUTH_REQUEST });
}
}
});
socket.on('disconnect', (code) => {
store.dispatch({ type: actions.DISCONNECTED, code });
});
})();
void (async () => {
for await (const data of socket.listener('disconnect')) {
store.dispatch({ type: actions.DISCONNECTED, code: data.code });
}
})();
socket.on('subscribe', (channel) => {
store.dispatch({ type: actions.SUBSCRIBE_SUCCESS, channel });
});
socket.on('unsubscribe', (channel) => {
socket.unsubscribe(channel);
socket.unwatch(channel);
socket.off(channel);
store.dispatch({ type: actions.UNSUBSCRIBE, channel });
});
socket.on('subscribeFail', (error) => {
store.dispatch({
type: actions.SUBSCRIBE_ERROR,
error,
status: 'subscribeFail',
});
});
socket.on('dropOut', (error) => {
store.dispatch({ type: actions.SUBSCRIBE_ERROR, error, status: 'dropOut' });
});
void (async () => {
for await (const data of socket.listener('subscribe')) {
store.dispatch({
type: actions.SUBSCRIBE_SUCCESS,
channel: data.channel,
});
}
})();
void (async () => {
for await (const data of socket.listener('unsubscribe')) {
socket.unsubscribe(data.channel);
store.dispatch({ type: actions.UNSUBSCRIBE, channel: data.channel });
}
})();
void (async () => {
for await (const data of socket.listener('subscribeFail')) {
store.dispatch({
type: actions.SUBSCRIBE_ERROR,
error: data.error,
status: 'subscribeFail',
});
}
})();
socket.on('error', (error) => {
store.dispatch({ type: actions.CONNECT_ERROR, error });
});
void (async () => {
for await (const data of socket.listener('error')) {
store.dispatch({ type: actions.CONNECT_ERROR, error: data.error });
}
})();
}
function connect() {
@ -205,43 +230,42 @@ function connect() {
function disconnect() {
if (socket) {
socket.disconnect();
socket.off();
}
}
function login() {
socket.emit('login', {}, (error: Error, baseChannel: string) => {
if (error) {
store.dispatch({ type: actions.AUTH_ERROR, error });
return;
void (async () => {
try {
const baseChannel = (await socket.invoke('login', {})) as string;
store.dispatch({ type: actions.AUTH_SUCCESS, baseChannel });
store.dispatch({
type: actions.SUBSCRIBE_REQUEST,
channel: baseChannel,
subscription: UPDATE_STATE,
});
store.dispatch({
type: actions.SUBSCRIBE_REQUEST,
channel: 'report',
subscription: UPDATE_REPORTS,
});
} catch (error) {
store.dispatch({ type: actions.AUTH_ERROR, error: error as Error });
}
store.dispatch({ type: actions.AUTH_SUCCESS, baseChannel });
store.dispatch({
type: actions.SUBSCRIBE_REQUEST,
channel: baseChannel,
subscription: UPDATE_STATE,
});
store.dispatch({
type: actions.SUBSCRIBE_REQUEST,
channel: 'report',
subscription: UPDATE_REPORTS,
});
});
})();
}
function getReport(reportId: unknown) {
socket.emit(
'getReport',
reportId,
(error: Error, data: { payload: string }) => {
if (error) {
store.dispatch({ type: GET_REPORT_ERROR, error });
return;
}
void (async () => {
try {
const data = (await socket.invoke('getReport', reportId)) as {
payload: string;
};
store.dispatch({ type: GET_REPORT_SUCCESS, data });
store.dispatch(importState(data.payload));
} catch (error) {
store.dispatch({ type: GET_REPORT_ERROR, error: error as Error });
}
);
})();
}
export function api(inStore: MiddlewareAPI<Dispatch<StoreAction>, StoreState>) {

View File

@ -71,7 +71,7 @@ export default function (argv: { [arg: string]: any }): Promise<{
const channel = action.receiver;
const data = action.data;
if (
channel.substr(0, 3) === 'sc-' ||
channel.substring(0, 3) === 'sc-' ||
channel === 'respond' ||
channel === 'log'
) {
@ -82,23 +82,21 @@ export default function (argv: { [arg: string]: any }): Promise<{
data: data,
});
}
} else if (action.type === action.SUBSCRIBE) {
if (action.channel === 'report') {
store
.list()
.then(function (data) {
void agServer.exchange.transmitPublish('report', {
type: 'list',
data: data,
});
})
.catch(function (error) {
console.error(error); // eslint-disable-line no-console
});
}
}
// TODO
// } else if (action.type === action.SUBSCRIBE) {
// if (action.channel === 'report') {
// store
// .list()
// .then(function (data) {
// action.socket.emit(action.channel, {
// type: 'list',
// data: data,
// });
// })
// .catch(function (error) {
// console.error(error); // eslint-disable-line no-console
// });
// }
// }
action.allow();
}
}
@ -117,12 +115,6 @@ export default function (argv: { [arg: string]: any }): Promise<{
channelToWatch = 'log';
channelToEmit = 'respond';
}
// TODO
// agServer.exchange
// .subscribe('sc-' + socket.id)
// .watch(function (msg) {
// socket.emit(channelToWatch, msg);
// });
request.end(channelToWatch);
}
})();
@ -141,11 +133,10 @@ export default function (argv: { [arg: string]: any }): Promise<{
})();
void (async () => {
for await (const data of socket.receiver('disconnect')) {
const channel = agServer.exchange.channel('sc-' + socket.id);
channel.unsubscribe();
// TODO
// const channel = agServer.exchange.channel('sc-' + socket.id);
// channel.unsubscribe();
// channel.destroy();
// agServer.exchange.publish(channelToEmit, {
// void agServer.exchange.transmitPublish(channelToEmit, {
// id: socket.id,
// type: 'DISCONNECTED',
// });

View File

@ -1,5 +1,5 @@
import { stringify, parse } from 'jsan';
import socketCluster, { SCClientSocket } from 'socketcluster-client';
import socketCluster, { AGClientSocket } from 'socketcluster-client';
import configureStore from './configureStore';
import { defaultSocketOptions } from './constants';
import getHostForRN from 'rn-host-detect';
@ -179,7 +179,7 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
store!: EnhancedStore<S, A, {}>;
filters: LocalFilter | undefined;
instanceId?: string;
socket?: SCClientSocket;
socket?: AGClientSocket;
sendTo?: string;
instanceName: string | undefined;
appInstanceId!: string;
@ -241,7 +241,8 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
) {
const message: MessageToRelay = {
type,
id: this.socket!.id,
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
id: this.socket!.id!,
name: this.instanceName,
instanceId: this.appInstanceId,
};
@ -279,7 +280,8 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
} else if (action) {
message.action = action as ActionCreatorObject[];
}
this.socket!.emit(this.socket!.id ? 'log' : 'log-noid', message);
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
this.socket!.transmit(this.socket!.id ? 'log' : 'log-noid', message);
}
dispatchRemotely(
@ -300,7 +302,9 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
if (
message.type === 'IMPORT' ||
(message.type === 'SYNC' &&
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
this.socket!.id &&
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
message.id !== this.socket!.id)
) {
this.store.liftedStore.dispatch({
@ -387,15 +391,22 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
}
login() {
this.socket!.emit('login', 'master', (err: Error, channelName: string) => {
if (err) {
console.log(err);
return;
void (async () => {
try {
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
const channelName = (await this.socket!.invoke(
'login',
'master'
)) as string;
this.channel = channelName;
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
for await (const data of this.socket!.subscribe(channelName)) {
this.handleMessages(data as Message<S, A>);
}
} catch (error) {
console.log(error);
}
this.channel = channelName;
this.socket!.subscribe(channelName).watch(this.handleMessages);
this.socket!.on(channelName, this.handleMessages);
});
})();
this.started = true;
this.relay('START');
}
@ -404,11 +415,9 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
this.started = false;
this.isMonitored = false;
if (!this.socket) return;
this.socket.destroyChannel(this.channel!);
if (keepConnected) {
this.socket.off(this.channel, this.handleMessages);
} else {
this.socket.off();
this.socket.unsubscribe(this.channel!);
this.socket.closeChannel(this.channel!);
if (!keepConnected) {
this.socket.disconnect();
}
};
@ -422,34 +431,46 @@ class DevToolsEnhancer<S, A extends Action<unknown>> {
this.socket = socketCluster.create(this.socketOptions);
this.socket.on('error', (err) => {
// if we've already had this error before, increment it's counter, otherwise assign it '1' since we've had the error once.
// eslint-disable-next-line no-prototype-builtins
this.errorCounts[err.name] = this.errorCounts.hasOwnProperty(err.name)
? this.errorCounts[err.name] + 1
: 1;
void (async () => {
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
for await (const data of this.socket!.listener('error')) {
// if we've already had this error before, increment it's counter, otherwise assign it '1' since we've had the error once.
// eslint-disable-next-line no-prototype-builtins,@typescript-eslint/no-unsafe-argument
this.errorCounts[data.error.name] = this.errorCounts.hasOwnProperty(
data.error.name
)
? this.errorCounts[data.error.name] + 1
: 1;
if (this.suppressConnectErrors) {
if (this.errorCounts[err.name] === 1) {
console.log(
'remote-redux-devtools: Socket connection errors are being suppressed. ' +
'\n' +
"This can be disabled by setting suppressConnectErrors to 'false'."
);
console.log(err);
if (this.suppressConnectErrors) {
if (this.errorCounts[data.error.name] === 1) {
console.log(
'remote-redux-devtools: Socket connection errors are being suppressed. ' +
'\n' +
"This can be disabled by setting suppressConnectErrors to 'false'."
);
console.log(data.error);
}
} else {
console.log(data.error);
}
} else {
console.log(err);
}
});
this.socket.on('connect', () => {
console.log('connected to remotedev-server');
this.errorCounts = {}; // clear the errorCounts object, so that we'll log any new errors in the event of a disconnect
this.login();
});
this.socket.on('disconnect', () => {
this.stop(true);
});
})();
void (async () => {
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
for await (const data of this.socket!.listener('connect')) {
console.log('connected to remotedev-server');
this.errorCounts = {}; // clear the errorCounts object, so that we'll log any new errors in the event of a disconnect
this.login();
}
})();
void (async () => {
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
for await (const data of this.socket!.listener('disconnect')) {
this.stop(true);
}
})();
};
checkForReducerErrors = (liftedState = this.getLiftedStateRaw()) => {