Fixed session/connection lateinit errors
This commit is contained in:
parent
8f9ee52b36
commit
7eac9699c9
|
@ -767,11 +767,7 @@ open class Client<CONNECTION : Connection>(config: ClientConfiguration = ClientC
|
|||
connectionInfo.secretKey
|
||||
))
|
||||
|
||||
|
||||
if (sessionManager.enabled()) {
|
||||
require(newConnection is SessionConnection) { "The new connection does not inherit a SessionConnection, unable to continue. " }
|
||||
}
|
||||
|
||||
sessionManager.onNewConnection(newConnection)
|
||||
|
||||
if (!handshakeConnection.pubSub.isIpc) {
|
||||
// NOTE: Client can ALWAYS connect to the server. The server makes the decision if the client can connect or not.
|
||||
|
@ -805,11 +801,8 @@ open class Client<CONNECTION : Connection>(config: ClientConfiguration = ClientC
|
|||
}
|
||||
|
||||
// in the specific case of using sessions, we don't want to call 'init' or `connect` for a connection that is resuming a session
|
||||
var newSession = true
|
||||
if (sessionManager.enabled()) {
|
||||
// we want to restore RMI objects BEFORE the connection is fully setup!
|
||||
newSession = sessionManager.onInit(newConnection as SessionConnection)
|
||||
}
|
||||
// when applicable - we ALSO want to restore RMI objects BEFORE the connection is fully setup!
|
||||
val newSession = sessionManager.onInit(newConnection)
|
||||
|
||||
newConnection.setImage()
|
||||
|
||||
|
|
|
@ -21,7 +21,7 @@ import kotlinx.atomicfu.locks.ReentrantLock
|
|||
import kotlinx.atomicfu.locks.withLock
|
||||
import java.util.concurrent.*
|
||||
|
||||
open class Session<CONNECTION: SessionConnection> {
|
||||
open class Session<CONNECTION: SessionConnection>(@Volatile var connection: CONNECTION) {
|
||||
|
||||
|
||||
// the RMI objects are saved when the connection is removed, and restored BEFORE the connection is initialized, so there are no concerns
|
||||
|
@ -36,7 +36,19 @@ open class Session<CONNECTION: SessionConnection> {
|
|||
*/
|
||||
val pendingMessagesQueue: LinkedTransferQueue<Any> = LinkedTransferQueue()
|
||||
|
||||
@Volatile lateinit var connection: CONNECTION
|
||||
|
||||
/**
|
||||
* the FIRST time this method is called, it will be true. EVERY SUBSEQUENT TIME, it will be false
|
||||
*/
|
||||
internal var isNewSession = true
|
||||
private set
|
||||
get() {
|
||||
val orig = field
|
||||
if (orig) {
|
||||
field = false
|
||||
}
|
||||
return orig
|
||||
}
|
||||
|
||||
|
||||
fun restore(connection: CONNECTION) {
|
||||
|
|
|
@ -21,14 +21,14 @@ import dorkbox.network.ClientConfiguration
|
|||
import dorkbox.network.connection.ConnectionParams
|
||||
|
||||
open class SessionClient<CONNECTION: SessionConnection>(config: ClientConfiguration = ClientConfiguration(), loggerName: String = Client::class.java.simpleName):
|
||||
Client<CONNECTION>(config, loggerName) {
|
||||
Client<CONNECTION>(config, loggerName), SessionEndpoint<CONNECTION> {
|
||||
|
||||
override fun newConnection(connectionParameters: ConnectionParams<CONNECTION>): CONNECTION {
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
return SessionConnection(connectionParameters) as CONNECTION
|
||||
}
|
||||
|
||||
open fun newSession(): Session<CONNECTION> {
|
||||
return Session()
|
||||
override fun newSession(connection: CONNECTION): Session<CONNECTION> {
|
||||
return Session(connection)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,24 @@
|
|||
/*
|
||||
* Copyright 2023 dorkbox, llc
|
||||
*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
package dorkbox.network.connection.session
|
||||
|
||||
import dorkbox.network.connection.ConnectionParams
|
||||
|
||||
interface SessionEndpoint<CONNECTION: SessionConnection> {
|
||||
fun newConnection(connectionParameters: ConnectionParams<CONNECTION>): CONNECTION
|
||||
fun newSession(connection: CONNECTION): Session<CONNECTION>
|
||||
}
|
|
@ -16,9 +16,12 @@
|
|||
|
||||
package dorkbox.network.connection.session
|
||||
|
||||
import dorkbox.network.connection.Connection
|
||||
|
||||
interface SessionManager<CONNECTION : SessionConnection> {
|
||||
|
||||
fun enabled(): Boolean
|
||||
fun onInit(connection: CONNECTION): Boolean
|
||||
fun onNewConnection(connection: Connection)
|
||||
fun onInit(connection: Connection): Boolean
|
||||
fun onDisconnect(connection: CONNECTION)
|
||||
}
|
||||
|
|
|
@ -21,6 +21,7 @@ import dorkbox.collections.LockFreeHashMap
|
|||
import dorkbox.hex.toHexString
|
||||
import dorkbox.network.Configuration
|
||||
import dorkbox.network.aeron.AeronDriver
|
||||
import dorkbox.network.connection.Connection
|
||||
import dorkbox.network.connection.EndPoint
|
||||
import dorkbox.util.Sys
|
||||
import net.jodah.expiringmap.ExpirationPolicy
|
||||
|
@ -73,45 +74,61 @@ internal open class SessionManagerFull<CONNECTION: SessionConnection>(
|
|||
return true
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* this must be called when a new connection is created AND when the internal `reconnect` occurs (as a result of a network error)
|
||||
* this must be called when a new connection is created
|
||||
*
|
||||
* @return true if this is a new session, false if it is an existing session
|
||||
*/
|
||||
override fun onInit(connection: CONNECTION): Boolean {
|
||||
override fun onNewConnection(connection: Connection) {
|
||||
require(connection is SessionConnection) { "The new connection does not inherit a SessionConnection, unable to continue. " }
|
||||
|
||||
val publicKeyWrapped = ByteArrayWrapper.wrap(connection.uuid)
|
||||
|
||||
var isNewSession = false
|
||||
val session = synchronized(sessions) {
|
||||
synchronized(sessions) {
|
||||
// always check if we are expiring first...
|
||||
val expiring = expiringSessions.remove(publicKeyWrapped)
|
||||
if (expiring != null) {
|
||||
// we must always set this session value!!
|
||||
connection.session = expiring
|
||||
expiring
|
||||
} else {
|
||||
val existing = sessions[publicKeyWrapped]
|
||||
if (existing != null) {
|
||||
// we must always set this session value!!
|
||||
connection.session = existing
|
||||
existing
|
||||
} else {
|
||||
isNewSession = true
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val newSession: Session<CONNECTION> = if (connection.endPoint.isServer()) {
|
||||
(connection.endPoint as SessionServer).newSession() as Session<CONNECTION>
|
||||
} else {
|
||||
(connection.endPoint as SessionClient).newSession() as Session<CONNECTION>
|
||||
}
|
||||
val newSession = (connection.endPoint as SessionEndpoint<CONNECTION>).newSession(connection as CONNECTION)
|
||||
|
||||
// we must always set this when the connection is created, and it must be inside the sync block!
|
||||
connection.session = newSession
|
||||
|
||||
sessions[publicKeyWrapped] = newSession
|
||||
newSession
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
connection.session = session
|
||||
session.restore(connection)
|
||||
|
||||
return isNewSession
|
||||
/**
|
||||
* this must be called when a new connection is created AND when the internal `reconnect` occurs (as a result of a network error)
|
||||
*
|
||||
* @return true if this is a new session, false if it is an existing session
|
||||
*/
|
||||
override fun onInit(connection: Connection): Boolean {
|
||||
// we know this will always be the case, because if this specific method can be called, then it will be a sessionConnection
|
||||
connection as SessionConnection
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val session: Session<CONNECTION> = connection.session as Session<CONNECTION>
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
session.restore(connection as CONNECTION)
|
||||
|
||||
// the FIRST time this method is called, it will be true. EVERY SUBSEQUENT TIME, it will be false
|
||||
return session.isNewSession
|
||||
}
|
||||
|
||||
|
||||
|
|
|
@ -16,12 +16,18 @@
|
|||
|
||||
package dorkbox.network.connection.session
|
||||
|
||||
import dorkbox.network.connection.Connection
|
||||
|
||||
class SessionManagerNoOp<CONNECTION : SessionConnection>: SessionManager<CONNECTION> {
|
||||
override fun enabled(): Boolean {
|
||||
return false
|
||||
}
|
||||
|
||||
override fun onInit(connection: CONNECTION): Boolean {
|
||||
override fun onNewConnection(connection: Connection) {
|
||||
// do nothing
|
||||
}
|
||||
|
||||
override fun onInit(connection: Connection): Boolean {
|
||||
// do nothing
|
||||
return true
|
||||
}
|
||||
|
|
|
@ -21,13 +21,13 @@ import dorkbox.network.ServerConfiguration
|
|||
import dorkbox.network.connection.ConnectionParams
|
||||
|
||||
open class SessionServer<CONNECTION: SessionConnection>(config: ServerConfiguration = ServerConfiguration(), loggerName: String = Server::class.java.simpleName):
|
||||
Server<CONNECTION>(config, loggerName) {
|
||||
Server<CONNECTION>(config, loggerName), SessionEndpoint<CONNECTION> {
|
||||
override fun newConnection(connectionParameters: ConnectionParams<CONNECTION>): CONNECTION {
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
return SessionConnection(connectionParameters) as CONNECTION
|
||||
}
|
||||
|
||||
open fun newSession(): Session<CONNECTION> {
|
||||
return Session()
|
||||
override fun newSession(connection: CONNECTION): Session<CONNECTION> {
|
||||
return Session(connection)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -134,11 +134,8 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
}
|
||||
|
||||
// in the specific case of using sessions, we don't want to call 'init' or `connect` for a connection that is resuming a session
|
||||
var newSession = true
|
||||
if (server.sessionManager.enabled()) {
|
||||
// we want to restore RMI objects BEFORE the connection is fully setup!
|
||||
newSession = server.sessionManager.onInit(newConnection as SessionConnection)
|
||||
}
|
||||
// when applicable - we ALSO want to restore RMI objects BEFORE the connection is fully setup!
|
||||
val newSession = server.sessionManager.onInit(newConnection)
|
||||
|
||||
newConnection.setImage()
|
||||
|
||||
|
@ -330,7 +327,7 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
|
||||
|
||||
// create a new connection. The session ID is encrypted.
|
||||
var connection: CONNECTION? = null
|
||||
var newConnection: CONNECTION? = null
|
||||
try {
|
||||
// Create a pub/sub at the given address and port, using the given stream ID.
|
||||
val newConnectionDriver = ServerConnectionDriver.build(
|
||||
|
@ -358,14 +355,17 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
logger.info("Creating new connection to $logInfo")
|
||||
}
|
||||
|
||||
connection = server.newConnection(ConnectionParams(
|
||||
publicKey,
|
||||
server,
|
||||
newConnectionDriver.pubSub,
|
||||
PublicKeyValidationState.VALID,
|
||||
CryptoManagement.NOCRYPT // we don't use encryption for IPC connections
|
||||
newConnection = server.newConnection(ConnectionParams(
|
||||
publicKey = publicKey,
|
||||
endPoint = server,
|
||||
connectionInfo = newConnectionDriver.pubSub,
|
||||
publicKeyValidation = PublicKeyValidationState.VALID,
|
||||
cryptoKey = CryptoManagement.NOCRYPT // we don't use encryption for IPC connections
|
||||
))
|
||||
|
||||
server.sessionManager.onNewConnection(newConnection)
|
||||
|
||||
|
||||
// VALIDATE:: are we allowed to connect to this server (now that we have the initial server information)
|
||||
// NOTE: all IPC client connections are, by default, always allowed to connect, because they are running on the same machine
|
||||
|
||||
|
@ -395,10 +395,10 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
successMessage.publicKey = server.crypto.publicKeyBytes
|
||||
|
||||
// before we notify connect, we have to wait for the client to tell us that they can receive data
|
||||
pendingConnections[message.connectKey] = connection
|
||||
pendingConnections[message.connectKey] = newConnection
|
||||
|
||||
if (logger.isDebugEnabled) {
|
||||
logger.debug("[$logInfo] (${message.connectKey}) Connection (${connection.id}) responding to handshake hello.")
|
||||
logger.debug("[$logInfo] (${message.connectKey}) Connection (${newConnection.id}) responding to handshake hello.")
|
||||
}
|
||||
|
||||
// this tells the client all the info to connect.
|
||||
|
@ -410,7 +410,7 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
streamIdAllocator.free(connectionStreamIdSub)
|
||||
streamIdAllocator.free(connectionStreamIdPub)
|
||||
|
||||
listenerManager.notifyError(ServerHandshakeException("[$logInfo] (${message.connectKey}) Connection (${connection?.id}) handshake crashed! Message $message", e))
|
||||
listenerManager.notifyError(ServerHandshakeException("[$logInfo] (${message.connectKey}) Connection (${newConnection?.id}) handshake crashed! Message $message", e))
|
||||
|
||||
return false
|
||||
}
|
||||
|
@ -558,7 +558,7 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
}
|
||||
|
||||
// create a new connection. The session ID is encrypted.
|
||||
var connection: CONNECTION? = null
|
||||
var newConnection: CONNECTION? = null
|
||||
try {
|
||||
// Create a pub/sub at the given address and port, using the given stream ID.
|
||||
val newConnectionDriver = ServerConnectionDriver.build(
|
||||
|
@ -589,13 +589,21 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
logger.info("Creating new connection to $logInfo")
|
||||
}
|
||||
|
||||
connection = server.newConnection(ConnectionParams(publicKey, server, newConnectionDriver.pubSub, validateRemoteAddress, cryptoSecretKey))
|
||||
newConnection = server.newConnection(ConnectionParams(
|
||||
publicKey = publicKey,
|
||||
endPoint = server,
|
||||
connectionInfo = newConnectionDriver.pubSub,
|
||||
publicKeyValidation = validateRemoteAddress,
|
||||
cryptoKey = cryptoSecretKey
|
||||
))
|
||||
|
||||
server.sessionManager.onNewConnection(newConnection)
|
||||
|
||||
// VALIDATE:: are we allowed to connect to this server (now that we have the initial server information)
|
||||
val permitConnection = listenerManager.notifyFilter(connection)
|
||||
val permitConnection = listenerManager.notifyFilter(newConnection)
|
||||
if (!permitConnection) {
|
||||
// this will also unwind/free allocations
|
||||
connection.close()
|
||||
newConnection.close()
|
||||
|
||||
listenerManager.notifyError(ServerHandshakeException("[$logInfo] Connection was not permitted!"))
|
||||
|
||||
|
@ -635,10 +643,10 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
successMessage.publicKey = server.crypto.publicKeyBytes
|
||||
|
||||
// before we notify connect, we have to wait for the client to tell us that they can receive data
|
||||
pendingConnections[message.connectKey] = connection
|
||||
pendingConnections[message.connectKey] = newConnection
|
||||
|
||||
if (logger.isDebugEnabled) {
|
||||
logger.debug("[$logInfo] (${message.connectKey}) Connection (${connection.id}) responding to handshake hello.")
|
||||
logger.debug("[$logInfo] (${message.connectKey}) Connection (${newConnection.id}) responding to handshake hello.")
|
||||
}
|
||||
|
||||
// this tells the client all the info to connect.
|
||||
|
@ -651,7 +659,7 @@ internal class ServerHandshake<CONNECTION : Connection>(
|
|||
streamIdAllocator.free(connectionStreamIdPub)
|
||||
streamIdAllocator.free(connectionStreamIdSub)
|
||||
|
||||
listenerManager.notifyError(ServerHandshakeException("[$logInfo] (${message.connectKey}) Connection (${connection?.id}) handshake crashed! Message $message", e))
|
||||
listenerManager.notifyError(ServerHandshakeException("[$logInfo] (${message.connectKey}) Connection (${newConnection?.id}) handshake crashed! Message $message", e))
|
||||
return false
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue