/**
* 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 hyperion.transport
/**
* 读消息事件处理器
*
* 支持在当前线程编解码和处理消息，或者将编解码和处理消息的任务交给业务线程池去执行
*
* @author yangfuping
*/
public abstract class ReadEventLoopHandler <: Runnable {
protected static let logger = LoggerFactory.getLogger("transport")
private let threadPoolVal: ThreadPool
protected let sessionVal: IoSession
private let allocator: ByteBufferAllocator
private var execOnReadThread = true
/**
* 能处理的最大消息长度，默认64M
*/
private var maxMessageSizeInBytes: Int64 = 64 * 1024 * 1024
public init(threadPool: ThreadPool, allocator: ByteBufferAllocator, session: IoSession) {
this.threadPoolVal = threadPool
this.allocator = allocator
this.sessionVal = session
}
public func setExecOnReadThread(execOnReadThread: Bool) {
this.execOnReadThread = execOnReadThread
}
public func setMaxMessageSizeInBytes(maxMessageSizeInBytes: Int64) {
if (maxMessageSizeInBytes > 0) {
this.maxMessageSizeInBytes = maxMessageSizeInBytes
}
}
public prop threadPool: ThreadPool {
get() {
return threadPoolVal
}
}
public prop session: IoSession {
get() {
return sessionVal
}
}
public func run() {
try {
doEventLoopProcess(session.connection)
} catch (ex: Exception) {
logger.log(LogLevel.ERROR, "Failure to process message on connection ${session.connection}", ex)
}
}
/**
* 读取消息
*/
private func doEventLoopProcess(connection: Connection) {
let status = MessageCompletedStatus()
var readCount = 0
var soTimeoutTimes = 0
var readCompletion: ?ReadCompletion = None
var readBuffer = allocator.allocate()
while (readCount != -1) {
try {
if (readBuffer.position() > 0) {
session.messageCompleted(readBuffer, status)
if (status.completed) {
var messageBuffer = readBuffer
if (let Some(messageSize) <- status.messageSize) {
messageBuffer = sliceMessageBuffer(messageSize, readBuffer)
// 重置MessageCompletedStatus
status.completed = false
status.messageSize = None
}
if (execOnReadThread) {
this.codecAndProcessMessage(session, messageBuffer)
} else {
// 交给业务线程池去执行编解码和消息的处理
let processTask = CodecAndProcessMessageHandler(session, messageBuffer,
this.codecAndProcessMessage)
threadPool.addTask(processTask)
}
continue
}
}
if (!readBuffer.hasRemaining()) {
// 扩展容量
readBuffer = expandBuffer(readBuffer)
}
// 尝试读取一些数据
readCount = connection.read(readBuffer)
soTimeoutTimes = 0
// 取消超时时间
connection.cancelExpireTime()
} catch (ex: SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.log(LogLevel.DEBUG, ex.message, ex)
}
if (!(readBuffer.position() > 0)) {
if (connection.isExpired()) {
// 连接空闲超时
logger.log(LogLevel.WARN, "Close idle timeout Connection ${connection}")
connection.close(markInvalid: true)
return
}
connection.setExpireTimeIfAbsent()
// 继续尝试读取
continue
} else {
soTimeoutTimes++
if (soTimeoutTimes > 3) {
logger.log(LogLevel.WARN, "Cloud not read data from ${connection}, close it")
connection.close(markInvalid: true)
return
}
// 读取到部分报文，再次尝试读取剩余报文
continue
}
} catch (ex: Exception) {
if (!connection.isClosed()) {
logger.log(LogLevel.ERROR, "Connection closed: ${connection}", ex)
connection.close(markInvalid: true)
} else {
if (logger.isDebugEnabled()) {
logger.log(LogLevel.DEBUG, "Connection already closed: ${connection}", ex)
}
}
onConnectionReadException(ex)
return
}
}
}
/**
* 分割出读取到的完整的消息报文，并将剩余部分设置到attachment中
*/
private func sliceMessageBuffer(messageSize: Int64, byteBuffer: ByteBuffer): ByteBuffer {
// 将已经读取的报文分割出来
let slice = byteBuffer.slice(index: 0, length: messageSize)
// 跳过已经读取到的报文
byteBuffer.skip(messageSize)
return slice
}
/**
* 扩展ByteBuffer
*
* @throws TransportException
*/
private func expandBuffer(byteBuffer: ByteBuffer): ByteBuffer {
if (byteBuffer.limit() > maxMessageSizeInBytes) {
throw TransportException(
"Too large message size: ${byteBuffer.limit()}, exceeding maxMessageSizeInBytes: ${maxMessageSizeInBytes}"
)
}
if (byteBuffer.expandSupport) {
var expandSize = byteBuffer.limit() << 1
if (expandSize > maxMessageSizeInBytes) {
expandSize = maxMessageSizeInBytes
}
byteBuffer.expandAndLimit(expandSize)
return byteBuffer
} else {
var expandSize = byteBuffer.limit() << 1
if (expandSize > maxMessageSizeInBytes) {
expandSize = maxMessageSizeInBytes
}
let expandBuffer = allocator.allocate(expandSize)
expandBuffer.put(byteBuffer.flip().toArray())
if (logger.isTraceEnabled()) {
logger.trace("Expand ${byteBuffer} to ${expandBuffer}")
}
// 释放扩容之前的ByteBuffer
byteBuffer.release()
return expandBuffer
}
}
public func onConnectionReadException(ex: Exception): Unit
/**
* 编解码和处理编解码后的消息
*/
private func codecAndProcessMessage(session: IoSession, messageBuffer: ByteBuffer): Unit {
/*
* 读完报文之后的回调，用于执行需要和读报文的顺序保持一致的操作。
* 例如： Redis的响应报文和发送的Redis命令的顺序需要保持一致。
*/
var readCompletion: ?ReadCompletion = None
let connection = session.connection
if (let Some(completionHandler) <- connection.getReadCompletionHandler()) {
readCompletion = completionHandler.readCompletion()
}
try {
if (let Some(readCompletion) <- readCompletion) {
let results = session.onMessageRead(messageBuffer, readCompletion)
if (!results.isEmpty()) {
processMessage(results)
}
} else {
let results = session.onMessageRead(messageBuffer)
if (!results.isEmpty()) {
processMessage(results)
}
}
} catch (ex: Exception) {
logger.log(LogLevel.ERROR, "Failure to process inbound message on ${connection}", ex)
} finally {
messageBuffer.release()
}
}
public func processMessage(messages: LinkedList<Any>): Unit
}
/**
* 只处理入栈消息
*/
public class StickyReadInboudHandler <: ReadEventLoopHandler {
public init(threadPool: ThreadPool, allocator: ByteBufferAllocator, session: IoSession) {
super(threadPool, allocator, session)
}
public func onConnectionReadException(ex: Exception): Unit {
session.onConnectionReadException(ex)
}
public func processMessage(messages: LinkedList<Any>): Unit {
// do nothing
}
}
/**
* 处理入栈消息，并将处理结果出栈
*/
public open class StickyReadInboudOutboundHandler <: ReadEventLoopHandler {
public init(threadPool: ThreadPool, allocator: ByteBufferAllocator, session: IoSession) {
super(threadPool, allocator, session)
}
public func onConnectionReadException(ex: Exception): Unit {
// do nothing
}
public func processMessage(messages: LinkedList<Any>): Unit {
for (message in messages) {
session.writeMessage(message)
}
session.flush()
}
}
/**
* 编解码和处理编解码后的消息
*/
public class CodecAndProcessMessageHandler <: Runnable {
protected static let logger = LoggerFactory.getLogger("transport")
private let session: IoSession
private let messageBuffer: ByteBuffer
private let handler: (session: IoSession, messageBuffer: ByteBuffer) -> Unit
public init(session: IoSession, messageBuffer: ByteBuffer,
handler: (session: IoSession, messageBuffer: ByteBuffer) -> Unit) {
this.session = session
this.messageBuffer = messageBuffer
this.handler = handler
}
public func run() {
handler(session, messageBuffer)
}
}