Adding some changes for to redis library, allowing reconnection.
This commit is contained in:
parent
0b728f07fa
commit
0a4c4f1cc0
|
@ -105,6 +105,8 @@ services:
|
||||||
restart: always
|
restart: always
|
||||||
image: redis
|
image: redis
|
||||||
command: redis-server --requirepass ${REDIS_PASSWORD}
|
command: redis-server --requirepass ${REDIS_PASSWORD}
|
||||||
|
ports:
|
||||||
|
- "${REDIS_PORT}:6379"
|
||||||
volumes:
|
volumes:
|
||||||
- redis_data:/data
|
- redis_data:/data
|
||||||
|
|
||||||
|
|
|
@ -3,22 +3,36 @@ const env = require("../environment")
|
||||||
const Redis = env.isTest() ? require("ioredis-mock") : require("ioredis")
|
const Redis = env.isTest() ? require("ioredis-mock") : require("ioredis")
|
||||||
const { addDbPrefix, removeDbPrefix, getRedisOptions } = require("./utils")
|
const { addDbPrefix, removeDbPrefix, getRedisOptions } = require("./utils")
|
||||||
|
|
||||||
|
const RETRY_PERIOD_MS = 2000
|
||||||
|
const MAX_RETRIES = 20
|
||||||
const CLUSTERED = false
|
const CLUSTERED = false
|
||||||
|
|
||||||
// for testing just generate the client once
|
// for testing just generate the client once
|
||||||
let CONNECTED = false
|
let CONNECTED = false
|
||||||
let CLIENT = env.isTest() ? new Redis(getRedisOptions()) : null
|
let CLIENT = env.isTest() ? new Redis(getRedisOptions()) : null
|
||||||
|
|
||||||
|
function retryConnection() {
|
||||||
|
setTimeout(init, RETRY_PERIOD_MS)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Inits the system, will error if unable to connect to redis cluster (may take up to 10 seconds) otherwise
|
* Inits the system, will error if unable to connect to redis cluster (may take up to 10 seconds) otherwise
|
||||||
* will return the ioredis client which will be ready to use.
|
* will return the ioredis client which will be ready to use.
|
||||||
* @return {Promise<object>} The ioredis client.
|
|
||||||
*/
|
*/
|
||||||
function init() {
|
function init() {
|
||||||
return new Promise((resolve, reject) => {
|
function errorOccurred(err) {
|
||||||
|
CONNECTED = false;
|
||||||
|
console.error("Redis connection failed - " + err)
|
||||||
|
setTimeout(() => {
|
||||||
|
init()
|
||||||
|
}, RETRY_PERIOD_MS)
|
||||||
|
}
|
||||||
// testing uses a single in memory client
|
// testing uses a single in memory client
|
||||||
if (env.isTest() || (CLIENT && CONNECTED)) {
|
if (env.isTest() || (CLIENT && CONNECTED)) {
|
||||||
return resolve(CLIENT)
|
return
|
||||||
|
}
|
||||||
|
if (CLIENT) {
|
||||||
|
CLIENT.disconnect()
|
||||||
}
|
}
|
||||||
const { opts, host, port } = getRedisOptions(CLUSTERED)
|
const { opts, host, port } = getRedisOptions(CLUSTERED)
|
||||||
if (CLUSTERED) {
|
if (CLUSTERED) {
|
||||||
|
@ -27,17 +41,27 @@ function init() {
|
||||||
CLIENT = new Redis(opts)
|
CLIENT = new Redis(opts)
|
||||||
}
|
}
|
||||||
CLIENT.on("end", err => {
|
CLIENT.on("end", err => {
|
||||||
reject(err)
|
errorOccurred(err)
|
||||||
CONNECTED = false
|
|
||||||
})
|
})
|
||||||
CLIENT.on("error", err => {
|
CLIENT.on("error", err => {
|
||||||
reject(err)
|
errorOccurred(err)
|
||||||
CONNECTED = false
|
|
||||||
})
|
})
|
||||||
CLIENT.on("connect", () => {
|
CLIENT.on("connect", () => {
|
||||||
resolve(CLIENT)
|
|
||||||
CONNECTED = true
|
CONNECTED = true
|
||||||
})
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
function waitForConnection() {
|
||||||
|
return new Promise(resolve => {
|
||||||
|
if (CLIENT == null) {
|
||||||
|
init()
|
||||||
|
} else if (CONNECTED) {
|
||||||
|
resolve()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
CLIENT.on("connect", () => {
|
||||||
|
resolve()
|
||||||
|
})
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -85,31 +109,30 @@ class RedisWrapper {
|
||||||
}
|
}
|
||||||
|
|
||||||
async init() {
|
async init() {
|
||||||
this._client = await init()
|
init()
|
||||||
|
await waitForConnection()
|
||||||
return this
|
return this
|
||||||
}
|
}
|
||||||
|
|
||||||
async finish() {
|
async finish() {
|
||||||
this._client.disconnect()
|
CLIENT.disconnect()
|
||||||
}
|
}
|
||||||
|
|
||||||
async scan() {
|
async scan() {
|
||||||
const db = this._db,
|
const db = this._db
|
||||||
client = this._client
|
|
||||||
let stream
|
let stream
|
||||||
if (CLUSTERED) {
|
if (CLUSTERED) {
|
||||||
let node = client.nodes("master")
|
let node = CLIENT.nodes("master")
|
||||||
stream = node[0].scanStream({ match: db + "-*", count: 100 })
|
stream = node[0].scanStream({ match: db + "-*", count: 100 })
|
||||||
} else {
|
} else {
|
||||||
stream = client.scanStream({ match: db + "-*", count: 100 })
|
stream = CLIENT.scanStream({ match: db + "-*", count: 100 })
|
||||||
}
|
}
|
||||||
return promisifyStream(stream)
|
return promisifyStream(stream)
|
||||||
}
|
}
|
||||||
|
|
||||||
async get(key) {
|
async get(key) {
|
||||||
const db = this._db,
|
const db = this._db
|
||||||
client = this._client
|
let response = await CLIENT.get(addDbPrefix(db, key))
|
||||||
let response = await client.get(addDbPrefix(db, key))
|
|
||||||
// overwrite the prefixed key
|
// overwrite the prefixed key
|
||||||
if (response != null && response.key) {
|
if (response != null && response.key) {
|
||||||
response.key = key
|
response.key = key
|
||||||
|
@ -123,22 +146,20 @@ class RedisWrapper {
|
||||||
}
|
}
|
||||||
|
|
||||||
async store(key, value, expirySeconds = null) {
|
async store(key, value, expirySeconds = null) {
|
||||||
const db = this._db,
|
const db = this._db
|
||||||
client = this._client
|
|
||||||
if (typeof value === "object") {
|
if (typeof value === "object") {
|
||||||
value = JSON.stringify(value)
|
value = JSON.stringify(value)
|
||||||
}
|
}
|
||||||
const prefixedKey = addDbPrefix(db, key)
|
const prefixedKey = addDbPrefix(db, key)
|
||||||
await client.set(prefixedKey, value)
|
await CLIENT.set(prefixedKey, value)
|
||||||
if (expirySeconds) {
|
if (expirySeconds) {
|
||||||
await client.expire(prefixedKey, expirySeconds)
|
await CLIENT.expire(prefixedKey, expirySeconds)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async delete(key) {
|
async delete(key) {
|
||||||
const db = this._db,
|
const db = this._db
|
||||||
client = this._client
|
await CLIENT.del(addDbPrefix(db, key))
|
||||||
await client.del(addDbPrefix(db, key))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async clear() {
|
async clear() {
|
||||||
|
|
|
@ -6,6 +6,13 @@ const Joi = require("joi")
|
||||||
|
|
||||||
const router = Router()
|
const router = Router()
|
||||||
|
|
||||||
|
function buildAdminInitValidation() {
|
||||||
|
return joiValidator.body(Joi.object({
|
||||||
|
email: Joi.string().required(),
|
||||||
|
password: Joi.string().required(),
|
||||||
|
}).required().unknown(false))
|
||||||
|
}
|
||||||
|
|
||||||
function buildUserSaveValidation(isSelf = false) {
|
function buildUserSaveValidation(isSelf = false) {
|
||||||
let schema = {
|
let schema = {
|
||||||
email: Joi.string().allow(null, ""),
|
email: Joi.string().allow(null, ""),
|
||||||
|
@ -74,7 +81,7 @@ router
|
||||||
buildInviteAcceptValidation(),
|
buildInviteAcceptValidation(),
|
||||||
controller.inviteAccept
|
controller.inviteAccept
|
||||||
)
|
)
|
||||||
.post("/api/admin/users/init", controller.adminUser)
|
.post("/api/admin/users/init", buildAdminInitValidation(), controller.adminUser)
|
||||||
.get("/api/admin/users/self", controller.getSelf)
|
.get("/api/admin/users/self", controller.getSelf)
|
||||||
// admin endpoint but needs to come at end (blocks other endpoints otherwise)
|
// admin endpoint but needs to come at end (blocks other endpoints otherwise)
|
||||||
.get("/api/admin/users/:id", adminOnly, controller.find)
|
.get("/api/admin/users/:id", adminOnly, controller.find)
|
||||||
|
|
Loading…
Reference in New Issue