mirror of https://github.com/calzoneman/sync.git
Remove junk from an old abandoned project
This commit is contained in:
Calvin Montgomery
parent
860775a90b
commit
00901f9cdb
|
|
@ -1,23 +0,0 @@
|
||||||
class BackendConfiguration {
|
|
||||||
constructor(config) {
|
|
||||||
this.config = config;
|
|
||||||
}
|
|
||||||
|
|
||||||
getRedisConfig() {
|
|
||||||
return this.config.redis;
|
|
||||||
}
|
|
||||||
|
|
||||||
getListenerConfig() {
|
|
||||||
return this.config.proxy.listeners.map(listener => ({
|
|
||||||
getHost() {
|
|
||||||
return listener.host;
|
|
||||||
},
|
|
||||||
|
|
||||||
getPort() {
|
|
||||||
return listener.port;
|
|
||||||
}
|
|
||||||
}));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export { BackendConfiguration };
|
|
||||||
|
|
@ -1,93 +0,0 @@
|
||||||
import { RedisClusterClient } from '../io/cluster/redisclusterclient';
|
|
||||||
import { DualClusterClient } from '../io/cluster/dualclusterclient';
|
|
||||||
import NullClusterClient from '../io/cluster/nullclusterclient';
|
|
||||||
import { FrontendPool } from 'cytube-common/lib/redis/frontendpool';
|
|
||||||
import RedisClientProvider from 'cytube-common/lib/redis/redisclientprovider';
|
|
||||||
import { loadFromToml } from 'cytube-common/lib/configuration/configloader';
|
|
||||||
import path from 'path';
|
|
||||||
import { BackendConfiguration } from './backendconfiguration';
|
|
||||||
import logger from 'cytube-common/lib/logger';
|
|
||||||
import redisAdapter from 'socket.io-redis';
|
|
||||||
import LegacyConfig from '../config';
|
|
||||||
import IOConfiguration from '../configuration/ioconfig';
|
|
||||||
import * as Switches from '../switches';
|
|
||||||
|
|
||||||
const BACKEND_CONFIG_PATH = path.resolve(__dirname, '..', '..', 'backend.toml');
|
|
||||||
|
|
||||||
class BackendModule {
|
|
||||||
constructor() {
|
|
||||||
this.initConfig();
|
|
||||||
}
|
|
||||||
|
|
||||||
initConfig() {
|
|
||||||
logger.initialize(null, null, LegacyConfig.get('debug'));
|
|
||||||
try {
|
|
||||||
this.backendConfig = loadFromToml(BackendConfiguration, BACKEND_CONFIG_PATH);
|
|
||||||
} catch (error) {
|
|
||||||
if (typeof error.line !== 'undefined') {
|
|
||||||
logger.error(`Error in configuration file: ${error} (line ${error.line})`);
|
|
||||||
} else {
|
|
||||||
logger.error(`Error loading configuration: ${error.stack}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
process.exit(1);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
onReady() {
|
|
||||||
const redisClientProvider = this.getRedisClientProvider();
|
|
||||||
this.redisAdapter = redisAdapter({
|
|
||||||
pubClient: redisClientProvider.get(),
|
|
||||||
// return_buffers is needed for msgpack-js to function correctly
|
|
||||||
subClient: redisClientProvider.get({ return_buffers: true })
|
|
||||||
});
|
|
||||||
this.sioEmitter = require('socket.io').instance;
|
|
||||||
this.sioEmitter.adapter(this.redisAdapter);
|
|
||||||
const IOBackend = require('./iobackend');
|
|
||||||
this.ioBackend = new IOBackend(
|
|
||||||
this.backendConfig.getListenerConfig()[0],
|
|
||||||
this.sioEmitter,
|
|
||||||
redisClientProvider.get()
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
getFrontendPool() {
|
|
||||||
if (!this.frontendPool) {
|
|
||||||
this.frontendPool = new FrontendPool(this.getRedisClientProvider().get());
|
|
||||||
}
|
|
||||||
|
|
||||||
return this.frontendPool;
|
|
||||||
}
|
|
||||||
|
|
||||||
getRedisClientProvider() {
|
|
||||||
if (!this.redisClientProvider) {
|
|
||||||
this.redisClientProvider = new RedisClientProvider(
|
|
||||||
this.backendConfig.getRedisConfig()
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
return this.redisClientProvider;
|
|
||||||
}
|
|
||||||
|
|
||||||
getClusterClient() {
|
|
||||||
if (!this.redisClusterClient) {
|
|
||||||
this.redisClusterClient = new RedisClusterClient(this.getFrontendPool());
|
|
||||||
}
|
|
||||||
|
|
||||||
if (Switches.isActive(Switches.DUAL_BACKEND) && !this.nullClusterClient) {
|
|
||||||
this.nullClusterClient = new NullClusterClient(
|
|
||||||
IOConfiguration.fromOldConfig(LegacyConfig));
|
|
||||||
}
|
|
||||||
|
|
||||||
if (Switches.isActive(Switches.DUAL_BACKEND)) {
|
|
||||||
this.clusterClient = new DualClusterClient(this.nullClusterClient,
|
|
||||||
this.redisClusterClient);
|
|
||||||
} else {
|
|
||||||
this.clusterClient = this.redisClusterClient;
|
|
||||||
}
|
|
||||||
|
|
||||||
return this.clusterClient;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export { BackendModule }
|
|
||||||
|
|
@ -1,45 +0,0 @@
|
||||||
import Server from 'cytube-common/lib/proxy/server';
|
|
||||||
import ProxyInterceptor from './proxyinterceptor';
|
|
||||||
import uuid from 'uuid';
|
|
||||||
import PoolEntryUpdater from 'cytube-common/lib/redis/poolentryupdater';
|
|
||||||
import JSONProtocol from 'cytube-common/lib/proxy/protocol';
|
|
||||||
import { formatProxyAddress } from 'cytube-common/lib/util/addressutil';
|
|
||||||
|
|
||||||
const BACKEND_POOL = 'backend-hosts';
|
|
||||||
|
|
||||||
export default class IOBackend {
|
|
||||||
constructor(proxyListenerConfig, socketEmitter, poolRedisClient) {
|
|
||||||
this.proxyListenerConfig = proxyListenerConfig;
|
|
||||||
this.socketEmitter = socketEmitter;
|
|
||||||
this.poolRedisClient = poolRedisClient;
|
|
||||||
this.protocol = new JSONProtocol();
|
|
||||||
this.initProxyInterceptor();
|
|
||||||
this.initProxyListener();
|
|
||||||
this.initBackendPoolUpdater();
|
|
||||||
}
|
|
||||||
|
|
||||||
initProxyInterceptor() {
|
|
||||||
this.proxyInterceptor = new ProxyInterceptor(this.socketEmitter);
|
|
||||||
}
|
|
||||||
|
|
||||||
initProxyListener() {
|
|
||||||
this.proxyListener = new Server(this.proxyListenerConfig, this.protocol);
|
|
||||||
this.proxyListener.on('connection',
|
|
||||||
this.proxyInterceptor.onConnection.bind(this.proxyInterceptor));
|
|
||||||
}
|
|
||||||
|
|
||||||
initBackendPoolUpdater() {
|
|
||||||
const hostname = this.proxyListenerConfig.getHost();
|
|
||||||
const port = this.proxyListenerConfig.getPort();
|
|
||||||
const entry = {
|
|
||||||
address: formatProxyAddress(hostname, port)
|
|
||||||
}
|
|
||||||
this.poolEntryUpdater = new PoolEntryUpdater(
|
|
||||||
this.poolRedisClient,
|
|
||||||
BACKEND_POOL,
|
|
||||||
uuid.v4(),
|
|
||||||
entry
|
|
||||||
);
|
|
||||||
this.poolEntryUpdater.start();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,54 +0,0 @@
|
||||||
import logger from 'cytube-common/lib/logger';
|
|
||||||
import { EventEmitter } from 'events';
|
|
||||||
|
|
||||||
export default class ProxiedSocket extends EventEmitter {
|
|
||||||
constructor(socketID, socketIP, socketUser, socketEmitter, frontendConnection) {
|
|
||||||
super();
|
|
||||||
this.id = socketID;
|
|
||||||
this.ip = socketIP;
|
|
||||||
this._realip = socketIP;
|
|
||||||
if (socketUser) {
|
|
||||||
this.user = {
|
|
||||||
name: socketUser.name,
|
|
||||||
global_rank: socketUser.globalRank
|
|
||||||
};
|
|
||||||
}
|
|
||||||
this.socketEmitter = socketEmitter;
|
|
||||||
this.frontendConnection = frontendConnection;
|
|
||||||
}
|
|
||||||
|
|
||||||
emit() {
|
|
||||||
const target = this.socketEmitter.to(this.id);
|
|
||||||
target.emit.apply(target, arguments);
|
|
||||||
}
|
|
||||||
|
|
||||||
onProxiedEventReceived() {
|
|
||||||
try {
|
|
||||||
EventEmitter.prototype.emit.apply(this, arguments);
|
|
||||||
} catch (error) {
|
|
||||||
logger.error(`Emit failed: ${error.stack}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
join(channel) {
|
|
||||||
this.frontendConnection.write(
|
|
||||||
this.frontendConnection.protocol.newSocketJoinRoomsEvent(
|
|
||||||
this.id, [channel]
|
|
||||||
)
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
leave(room) {
|
|
||||||
this.frontendConnection.write(
|
|
||||||
this.frontendConnection.protocol.newSocketLeaveRoomsEvent(
|
|
||||||
this.id, [room]
|
|
||||||
)
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
disconnect() {
|
|
||||||
this.frontendConnection.write(
|
|
||||||
this.frontendConnection.protocol.newSocketKickEvent(this.id)
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,91 +0,0 @@
|
||||||
import logger from 'cytube-common/lib/logger';
|
|
||||||
import ioServer from '../io/ioserver';
|
|
||||||
import ProxiedSocket from './proxiedsocket';
|
|
||||||
|
|
||||||
export default class ProxyInterceptor {
|
|
||||||
constructor(socketEmitter) {
|
|
||||||
this.socketEmitter = socketEmitter;
|
|
||||||
this.frontendConnections = {};
|
|
||||||
this.frontendProxiedSockets = {};
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Handle a new frontend proxy connection.
|
|
||||||
*
|
|
||||||
* @param {Connection} socket frontend proxy connection
|
|
||||||
*/
|
|
||||||
onConnection(socket) {
|
|
||||||
if (this.frontendConnections.hasOwnProperty(socket.endpoint)) {
|
|
||||||
logger.error(`Duplicate frontend connection: ${socket.endpoint}`);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.info(`Got proxy connection from ${socket.endpoint}`);
|
|
||||||
this.frontendConnections[socket.endpoint] = socket;
|
|
||||||
socket.on('close', this.onFrontendDisconnect.bind(this, socket));
|
|
||||||
socket.on('SocketConnectEvent', this.onSocketConnect.bind(this, socket));
|
|
||||||
socket.on('SocketFrameEvent', this.onSocketFrame.bind(this, socket));
|
|
||||||
socket.on('SocketDisconnectEvent', this.onSocketDisconnect.bind(this, socket));
|
|
||||||
}
|
|
||||||
|
|
||||||
onFrontendDisconnect(socket) {
|
|
||||||
const endpoint = socket.endpoint;
|
|
||||||
if (this.frontendConnections.hasOwnProperty(endpoint)) {
|
|
||||||
if (this.frontendProxiedSockets.hasOwnProperty(endpoint)) {
|
|
||||||
logger.warn(`Frontend ${endpoint} disconnected`);
|
|
||||||
for (const key in this.frontendProxiedSockets[endpoint]) {
|
|
||||||
const proxySocket = this.frontendProxiedSockets[endpoint][key];
|
|
||||||
proxySocket.onProxiedEventReceived('disconnect');
|
|
||||||
}
|
|
||||||
delete this.frontendProxiedSockets[endpoint];
|
|
||||||
}
|
|
||||||
delete this.frontendConnections[endpoint];
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
onSocketConnect(frontendConnection, socketID, socketIP, socketUser) {
|
|
||||||
const mapKey = frontendConnection.endpoint;
|
|
||||||
const proxiedSocket = new ProxiedSocket(
|
|
||||||
socketID,
|
|
||||||
socketIP,
|
|
||||||
socketUser,
|
|
||||||
this.socketEmitter,
|
|
||||||
frontendConnection);
|
|
||||||
|
|
||||||
if (!this.frontendProxiedSockets.hasOwnProperty(mapKey)) {
|
|
||||||
this.frontendProxiedSockets[mapKey] = {};
|
|
||||||
} else if (this.frontendProxiedSockets[mapKey].hasOwnProperty(socketID)) {
|
|
||||||
logger.error(`Duplicate SocketConnectEvent for ${socketID}`);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
this.frontendProxiedSockets[mapKey][socketID] = proxiedSocket;
|
|
||||||
ioServer.handleConnection(proxiedSocket);
|
|
||||||
}
|
|
||||||
|
|
||||||
onSocketFrame(frontendConnection, socketID, event, args) {
|
|
||||||
const mapKey = frontendConnection.endpoint;
|
|
||||||
const socketMap = this.frontendProxiedSockets[mapKey];
|
|
||||||
if (!socketMap || !socketMap.hasOwnProperty(socketID)) {
|
|
||||||
logger.error(`Received SocketFrameEvent for nonexistent socket`,
|
|
||||||
{ socketID, event });
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
const socket = socketMap[socketID];
|
|
||||||
socket.onProxiedEventReceived.apply(socket, [event].concat(args));
|
|
||||||
}
|
|
||||||
|
|
||||||
onSocketDisconnect(frontendConnection, socketID) {
|
|
||||||
const mapKey = frontendConnection.endpoint;
|
|
||||||
const socketMap = this.frontendProxiedSockets[mapKey];
|
|
||||||
if (!socketMap || !socketMap.hasOwnProperty(socketID)) {
|
|
||||||
logger.error(`Received SocketDisconnectEvent for nonexistent socket`,
|
|
||||||
{ socketID });
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
const socket = socketMap[socketID];
|
|
||||||
socket.onProxiedEventReceived.apply(socket, ['disconnect']);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,27 +0,0 @@
|
||||||
import logger from 'cytube-common/lib/logger';
|
|
||||||
import * as Switches from '../../switches';
|
|
||||||
|
|
||||||
class DualClusterClient {
|
|
||||||
constructor(authoritativeClient, altClient) {
|
|
||||||
this.authoritativeClient = authoritativeClient;
|
|
||||||
this.altClient = altClient;
|
|
||||||
}
|
|
||||||
|
|
||||||
getSocketConfig(channel) {
|
|
||||||
return this.authoritativeClient.getSocketConfig(channel).then(result => {
|
|
||||||
if (!Switches.isActive(Switches.DUAL_BACKEND)) {
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
|
|
||||||
return this.altClient.getSocketConfig(channel).then(altResult => {
|
|
||||||
result.alt = altResult.servers;
|
|
||||||
return result;
|
|
||||||
}).catch(error => {
|
|
||||||
logger.warn(`Error loading alt servers: ${error}`);
|
|
||||||
return result;
|
|
||||||
});
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export { DualClusterClient };
|
|
||||||
|
|
@ -1,22 +0,0 @@
|
||||||
import Promise from 'bluebird';
|
|
||||||
|
|
||||||
const ONE_SECOND = 1000;
|
|
||||||
const ERR_TIMEOUT = 'Timed out when retrieving server information';
|
|
||||||
|
|
||||||
class RedisClusterClient {
|
|
||||||
constructor(frontendPool) {
|
|
||||||
this.frontendPool = frontendPool;
|
|
||||||
}
|
|
||||||
|
|
||||||
getSocketConfig(channel) {
|
|
||||||
return this.frontendPool.getFrontends(channel).then(result => {
|
|
||||||
if (!Array.isArray(result)) {
|
|
||||||
result = [];
|
|
||||||
}
|
|
||||||
|
|
||||||
return { servers: result };
|
|
||||||
}).timeout(ONE_SECOND, ERR_TIMEOUT);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export { RedisClusterClient };
|
|
||||||
|
|
@ -64,15 +64,8 @@ var Server = function () {
|
||||||
self.servers = {};
|
self.servers = {};
|
||||||
self.chanPath = Config.get('channel-path');
|
self.chanPath = Config.get('channel-path');
|
||||||
|
|
||||||
// backend init
|
|
||||||
var initModule;
|
var initModule;
|
||||||
if (Config.get("new-backend")) {
|
if (Config.get('enable-partition')) {
|
||||||
if (Config.get("dual-backend")) {
|
|
||||||
Switches.setActive(Switches.DUAL_BACKEND, true);
|
|
||||||
}
|
|
||||||
const BackendModule = require('./backend/backendmodule').BackendModule;
|
|
||||||
initModule = this.initModule = new BackendModule();
|
|
||||||
} else if (Config.get('enable-partition')) {
|
|
||||||
initModule = this.initModule = new PartitionModule();
|
initModule = this.initModule = new PartitionModule();
|
||||||
self.partitionDecider = initModule.getPartitionDecider();
|
self.partitionDecider = initModule.getPartitionDecider();
|
||||||
} else {
|
} else {
|
||||||
|
|
|
||||||
|
|
@ -7,5 +7,3 @@ export function isActive(switchName) {
|
||||||
export function setActive(switchName, active) {
|
export function setActive(switchName, active) {
|
||||||
switches[switchName] = active;
|
switches[switchName] = active;
|
||||||
}
|
}
|
||||||
|
|
||||||
export const DUAL_BACKEND = 'DUAL_BACKEND';
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue