-
Notifications
You must be signed in to change notification settings - Fork 822
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement and use a Session data storage (#721)
Introduce the concept of SessionsRepository to store session's data that are not subscriptions or queues; those are already persisted with their repository instances. This is a step to move to MQTT5 which benefit also the MQTT3 implementation. introduce ISessionRepository interface and its H2 implementation the session data are stored in a new data class named SessionData implements serializitaion/deserialization of SessionData for H2 * Capped the infinite session expire interval to 100 years (as seconds instead of UINT max value) * Replaces session semi-final fields in Session live object with SessionData
- Loading branch information
Showing
16 changed files
with
295 additions
and
35 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
96 changes: 96 additions & 0 deletions
96
broker/src/main/java/io/moquette/broker/ISessionsRepository.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| package io.moquette.broker; | ||
|
|
||
| import io.netty.handler.codec.mqtt.MqttVersion; | ||
|
|
||
| import java.time.Instant; | ||
| import java.util.Collection; | ||
| import java.util.Objects; | ||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * Used to store data about persisted sessions like MQTT version, session's properties. | ||
| * */ | ||
| public interface ISessionsRepository { | ||
|
|
||
| // Data class | ||
| final class SessionData { | ||
| private final String clientId; | ||
| private Instant expireAt = null; | ||
| final MqttVersion version; | ||
| private final int expiryInterval; | ||
|
|
||
| /** | ||
| * Construct a new SessionData without expiration set yet. | ||
| * */ | ||
| public SessionData(String clientId, MqttVersion version, int expiryInterval) { | ||
| this.clientId = clientId; | ||
| this.version = version; | ||
| this.expiryInterval = expiryInterval; | ||
| } | ||
|
|
||
| /** | ||
| * Construct SessionData with an expiration instant, created by loading from the storage. | ||
| * */ | ||
| public SessionData(String clientId, Instant expireAt, MqttVersion version, int expiryInterval) { | ||
| this.expiryInterval = expiryInterval; | ||
| Objects.requireNonNull(expireAt, "An expiration time is requested"); | ||
| this.clientId = clientId; | ||
| this.expireAt = expireAt; | ||
| this.version = version; | ||
| } | ||
|
|
||
| public String clientId() { | ||
| return clientId; | ||
| } | ||
|
|
||
| public MqttVersion protocolVersion() { | ||
| return version; | ||
| } | ||
|
|
||
| public Optional<Instant> expireAt() { | ||
| return Optional.ofNullable(expireAt); | ||
| } | ||
|
|
||
| public Optional<Long> expiryInstant() { | ||
| return expireAt() | ||
| .map(Instant::toEpochMilli); | ||
| } | ||
|
|
||
| public int expiryInterval() { | ||
| return expiryInterval; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object o) { | ||
| if (this == o) return true; | ||
| if (o == null || getClass() != o.getClass()) return false; | ||
| SessionData that = (SessionData) o; | ||
| return clientId.equals(that.clientId); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(clientId); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "SessionData{" + | ||
| "clientId='" + clientId + '\'' + | ||
| ", expireAt=" + expireAt + | ||
| ", version=" + version + | ||
| ", expiryInterval=" + expiryInterval + | ||
| '}'; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @return the full list of persisted sessions data. | ||
| * */ | ||
| Collection<SessionData> list(); | ||
|
|
||
| /** | ||
| * Save data composing a session, es MQTT version, creation date and properties but not queues or subscriptions. | ||
| * */ | ||
| void saveSession(SessionData session); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.