/**
* Copyright 2024 Beijing Baolande Software Corporation
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Runtime Library Exception to the Apache 2.0 License:
*
* As an exception, if you use this Software to compile your source code and
* portions of this Software are embedded into the binary product as a result,
* you may redistribute such product without providing attribution as would
* otherwise be required by Sections 4(a), 4(b) and 4(d) of the License.
*/
package redis_sdk.client
/**
* 处理订阅模式请求发送和响应消息推送
*
* 由于订阅模式不是一命令一响应的模式，不能使用和处理常规命令相同的IoFilter实现：
* 连接未切换成订阅模式时，采用和普通命令的一命令一响应的处理方式，收到响应消息后设置给对应的RedisCommand；
* 连接切换为订阅模式后，不再向RedisCommand设置响应消息，将响应消息放入响应消息队列中，后续由SubscriberEventLoop取走处理；
*/
public class RedisSubscriberHandler <: SingularMessageIoFilter {
private static let instance = RedisSubscriberHandler()
public static func getInstance(): RedisSubscriberHandler {
return instance
}
// 每个连接一个命令队列
private let commandsQueueMap = ConcurrentHashMap<Connection, LinkedBlockingQueue<RedisCommand>>()
// 每个连接一个响应队列, 用于处理订阅模式的命令
private let responseQueueMap = ConcurrentHashMap<Connection, LinkedBlockingQueue<SubscribeResponse>>()
public init() {
// 出现异常时不能跳过此IoFilter
super(skipOnException: false)
}
/**
* 处理入栈消息
*
* @throws TransportException
*/
public func processInboundMessage(context: IoFilterContext, session: Session, inMessage: Any): Unit {
if (let Some(responseQueue) <- getResponseQueue(session.connection)) {
// 订阅模式
let subscribeResponse = SubscribeResponse()
if (let Some(exception) <- context.getException()) {
subscribeResponse.exception = exception
responseQueue.add(subscribeResponse)
return
}
if (let Some(redisMessage) <- (inMessage as RedisMessage)) {
// 订阅模式时，不能忽略Pushes
subscribeResponse.redisMessage = redisMessage
responseQueue.add(subscribeResponse)
return
}
let exception = TransportException("Unsupported message, only accept RedisMessage")
subscribeResponse.exception = exception
responseQueue.add(subscribeResponse)
} else {
let commandsQueue = getCommandsQueue(session.connection)
let command = commandsQueue.remove()
if (let Some(exception) <- context.getException()) {
command.setException(exception)
return
}
if (let Some(redisMessage) <- (inMessage as RedisMessage)) {
if (let Some(redisMessage) <- (redisMessage as PushesRedisMessage)) {
// 非订阅模式忽略Pushes
return
}
command.setResponse(redisMessage)
return
}
let exception = TransportException("Unsupported message, only accept RedisMessage")
command.setException(exception)
}
}
/**
* 处理入栈异常
*
*/
public open func processInboundException(context: IoFilterContext, session: Session, ex: Exception): Unit {
if (let Some(responseQueue) <- getResponseQueue(session.connection)) {
// 订阅模式
let subscribeResponse = SubscribeResponse()
subscribeResponse.exception = ex
responseQueue.add(subscribeResponse)
} else {
let commandsQueue = getCommandsQueue(session.connection)
while (let Some(command) <- commandsQueue.tryRemove()) {
// 主动关闭连接导致读数据异常的场景，命令队列中不一定存在RedisCommand
command.setException(ex)
}
}
}
/**
* 处理出栈消息
*
* @throws TransportException
*/
public func processOutboundMessage(context: IoFilterContext, session: Session, outMessage: Any): Unit {
if (context.isExceptionCaughted()) {
return
}
if (let Some(redisCommand) <- (outMessage as RedisCommand)) {
context.offerMessage(redisCommand)
if (getResponseQueue(session.connection).isNone()) {
// 非订阅模式，入栈命令
let commandsQueue = getCommandsQueue(session.connection)
commandsQueue.add(redisCommand)
}
return
}
let exception = TransportException("Unsupported message, only accept RedisCommand")
context.exceptionCaught(exception)
}
/**
* 处理出栈异常
*
*/
public func processOutboundException(context: IoFilterContext, session: Session, ex: Exception): Unit {
context.exceptionCaught(ex)
}
/**
* 将连接的请求处理切换到订阅模式
*
*/
public func switchToSubscriberMode(session: Session): Unit {
let conn = session.connection
var responseQueue = responseQueueMap.get(conn)
if (let Some(responseQueue) <- responseQueue) {
// 该连接的响应处理已经转换为订阅模式
return
}
let craetedResponseQueue = LinkedBlockingQueue<SubscribeResponse>()
responseQueue = responseQueueMap.addIfAbsent(conn, craetedResponseQueue)
if (responseQueue.isNone()) {
let removeListener = RomoveResponseQueueLister(conn, this)
conn.addListener(removeListener)
// 将读取超时时间设置为无穷大
conn.setTimeoutInfinite()
}
}
/**
* 获取连接对应的响应队列
*/
private func getResponseQueue(conn: Connection): ?LinkedBlockingQueue<SubscribeResponse> {
let responseQueue = responseQueueMap.get(conn)
if (let Some(responseQueue) <- responseQueue) {
return responseQueue
}
return None
}
/**
* 从连接的响应队列中弹出一个响应
*/
public func popSubscribeResponse(session: Session, waitTime: Duration): ?SubscribeResponse {
try {
if (let Some(responseQueue) <- getResponseQueue(session.connection)) {
return responseQueue.remove(waitTime)
}
throw RedisException("Not in subscribe mode")
} catch (ex: Exception) {
if (let Some(redisEx) <- (ex as RedisException)) {
throw redisEx
}
let redisEx = RedisException(ex.message, ex)
throw redisEx
}
}
public func removeResponseQueue(conn: Connection) {
responseQueueMap.remove(conn)
// 还原读取超时时间
conn.rollbackTimeout()
}
/**
* 获取连接对应的命令队列
*/
private func getCommandsQueue(conn: Connection): LinkedBlockingQueue<RedisCommand> {
var commandsQueue = commandsQueueMap.get(conn)
if (let Some(commandsQueue) <- commandsQueue) {
return commandsQueue
}
let createdCommandsQueue = LinkedBlockingQueue<RedisCommand>()
commandsQueue = commandsQueueMap.addIfAbsent(conn, createdCommandsQueue)
if (let Some(commandsQueue) <- commandsQueue) {
return commandsQueue
} else {
if (!conn.isClosed()) {
let removeListener = RomoveResponseQueueLister(conn, this)
conn.addListener(removeListener)
}
return createdCommandsQueue
}
}
public func removeCommandsQueue(conn: Connection) {
commandsQueueMap.remove(conn)
}
public func toString(): String {
return "RedisSubscriberHandler"
}
}
public class SubscribeResponse {
private var redisMessageVal: ?RedisMessage = None
private var exceptionVal: ?Exception = None
public mut prop redisMessage: ?RedisMessage {
get() {
return redisMessageVal
}
set(val) {
this.redisMessageVal = val
}
}
public mut prop exception: ?Exception {
get() {
return exceptionVal
}
set(val) {
this.exceptionVal = val
}
}
}
class RomoveResponseQueueLister <: ConnectionLister {
private let conn: Connection
private let redisPubsubHandler: RedisSubscriberHandler
public init(conn: Connection, redisPubsubHandler: RedisSubscriberHandler) {
this.conn = conn
this.redisPubsubHandler = redisPubsubHandler
}
public func onClose(): Unit {
spawn {
=>
// 延迟一会儿再清理
sleep(Duration.millisecond * 500)
redisPubsubHandler.removeResponseQueue(conn)
redisPubsubHandler.removeCommandsQueue(conn)
}
}
}