From 4dc602d8a7700924b0117424533046b28f4a8df4 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Mon, 30 May 2022 18:39:50 -0700 Subject: Combine active device and session --- .../java/org/traccar/session/ActiveDevice.java | 58 ---------------- .../org/traccar/session/ConnectionManager.java | 81 +++++++++++++++++++--- .../java/org/traccar/session/DeviceSession.java | 40 ++++++++++- src/main/java/org/traccar/session/Endpoint.java | 58 ++++++++++++++++ 4 files changed, 166 insertions(+), 71 deletions(-) delete mode 100644 src/main/java/org/traccar/session/ActiveDevice.java create mode 100644 src/main/java/org/traccar/session/Endpoint.java (limited to 'src/main/java/org/traccar/session') diff --git a/src/main/java/org/traccar/session/ActiveDevice.java b/src/main/java/org/traccar/session/ActiveDevice.java deleted file mode 100644 index af19ba55b..000000000 --- a/src/main/java/org/traccar/session/ActiveDevice.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Copyright 2015 - 2020 Anton Tananaev (anton@traccar.org) - * - * 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 org.traccar.session; - -import io.netty.channel.Channel; -import io.netty.handler.codec.http.HttpRequestDecoder; -import org.traccar.BasePipelineFactory; -import org.traccar.Protocol; -import org.traccar.model.Command; - -import java.net.SocketAddress; - -public class ActiveDevice { - - private final long deviceId; - private final Protocol protocol; - private final Channel channel; - private final SocketAddress remoteAddress; - private final boolean supportsLiveCommands; - - public ActiveDevice(long deviceId, Protocol protocol, Channel channel, SocketAddress remoteAddress) { - this.deviceId = deviceId; - this.protocol = protocol; - this.channel = channel; - this.remoteAddress = remoteAddress; - supportsLiveCommands = BasePipelineFactory.getHandler(channel.pipeline(), HttpRequestDecoder.class) == null; - } - - public Channel getChannel() { - return channel; - } - - public long getDeviceId() { - return deviceId; - } - - public boolean supportsLiveCommands() { - return supportsLiveCommands; - } - - public void sendCommand(Command command) { - protocol.sendDataCommand(channel, remoteAddress, command); - } - -} diff --git a/src/main/java/org/traccar/session/ConnectionManager.java b/src/main/java/org/traccar/session/ConnectionManager.java index fbc15b00d..5d8a8c606 100644 --- a/src/main/java/org/traccar/session/ConnectionManager.java +++ b/src/main/java/org/traccar/session/ConnectionManager.java @@ -31,6 +31,7 @@ import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.storage.StorageException; +import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.Date; import java.util.HashMap; @@ -47,7 +48,9 @@ public class ConnectionManager { private final long deviceTimeout; private final boolean updateDeviceState; - private final Map activeDevices = new ConcurrentHashMap<>(); + private final Map sessionsByDeviceId = new ConcurrentHashMap<>(); + private final Map> sessionsByEndpoint = new ConcurrentHashMap<>(); + private final Map> listeners = new ConcurrentHashMap<>(); private final Map timeouts = new ConcurrentHashMap<>(); @@ -59,22 +62,78 @@ public class ConnectionManager { timer = Main.getInjector().getInstance(Timer.class); } - public void addActiveDevice(long deviceId, Protocol protocol, Channel channel, SocketAddress remoteAddress) { - activeDevices.put(deviceId, new ActiveDevice(deviceId, protocol, channel, remoteAddress)); + public DeviceSession getDeviceSession(long deviceId) { + return sessionsByDeviceId.get(deviceId); } - public void removeActiveDevice(Channel channel) { - for (ActiveDevice activeDevice : activeDevices.values()) { - if (activeDevice.getChannel() == channel) { - updateDevice(activeDevice.getDeviceId(), Device.STATUS_OFFLINE, null); - activeDevices.remove(activeDevice.getDeviceId()); - break; + public DeviceSession getDeviceSession( + Protocol protocol, Channel channel, SocketAddress remoteAddress, String... uniqueIds) { + + Endpoint endpoint = new Endpoint(channel, remoteAddress); + Map endpointSessions = sessionsByEndpoint.getOrDefault( + endpoint, new ConcurrentHashMap<>()); + if (uniqueIds.length > 0) { + for (String uniqueId : uniqueIds) { + DeviceSession deviceSession = endpointSessions.get(uniqueId); + if (deviceSession != null) { + return deviceSession; + } } + } else { + return endpointSessions.values().stream().findAny().orElse(null); + } + + Device device = null; + try { + for (String uniqueId : uniqueIds) { + device = Context.getIdentityManager().getByUniqueId(uniqueId); + if (device != null) { + break; + } + } + } catch (Exception e) { + LOGGER.warn("Find device error", e); + } + + if (device == null && Context.getConfig().getBoolean(Keys.DATABASE_REGISTER_UNKNOWN)) { + device = Context.getIdentityManager().addUnknownDevice(uniqueIds[0]); + } + + if (device != null && !device.getDisabled()) { + DeviceSession oldSession = sessionsByDeviceId.remove(device.getId()); + if (oldSession != null) { + Endpoint oldEndpoint = new Endpoint(oldSession.getChannel(), oldSession.getRemoteAddress()); + Map oldEndpointSessions = sessionsByEndpoint.get(oldEndpoint); + if (oldEndpointSessions.size() > 1) { + oldEndpointSessions.remove(device.getUniqueId()); + } else { + sessionsByEndpoint.remove(oldEndpoint); + } + } + + DeviceSession deviceSession = new DeviceSession( + device.getId(), device.getUniqueId(), protocol, channel, remoteAddress); + endpointSessions.put(device.getUniqueId(), deviceSession); + sessionsByEndpoint.put(endpoint, endpointSessions); + sessionsByDeviceId.put(device.getId(), deviceSession); + + return deviceSession; + } else { + LOGGER.warn((device == null ? "Unknown" : "Disabled") + " device - " + String.join(" ", uniqueIds) + + " (" + ((InetSocketAddress) remoteAddress).getHostString() + ")"); + return null; } } - public ActiveDevice getActiveDevice(long deviceId) { - return activeDevices.get(deviceId); + public void removeDeviceSessions(Channel channel) { + Endpoint endpoint = new Endpoint(channel, channel.remoteAddress()); + Map endpointSessions = sessionsByEndpoint.remove(endpoint); + if (endpointSessions != null) { + for (DeviceSession deviceSession : endpointSessions.values()) { + updateDevice(deviceSession.getDeviceId(), Device.STATUS_OFFLINE, null); + sessionsByDeviceId.remove(deviceSession.getDeviceId()); + } + } } public void updateDevice(final long deviceId, String status, Date time) { diff --git a/src/main/java/org/traccar/session/DeviceSession.java b/src/main/java/org/traccar/session/DeviceSession.java index 0d5b283fe..009f90f5a 100644 --- a/src/main/java/org/traccar/session/DeviceSession.java +++ b/src/main/java/org/traccar/session/DeviceSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2018 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -15,21 +15,57 @@ */ package org.traccar.session; +import io.netty.channel.Channel; +import io.netty.handler.codec.http.HttpRequestDecoder; +import org.traccar.BasePipelineFactory; +import org.traccar.Protocol; +import org.traccar.model.Command; + +import java.net.SocketAddress; import java.util.HashMap; import java.util.Map; public class DeviceSession { private final long deviceId; + private final String uniqueId; + private final Protocol protocol; + private final Channel channel; + private final SocketAddress remoteAddress; - public DeviceSession(long deviceId) { + public DeviceSession( + long deviceId, String uniqueId, Protocol protocol, Channel channel, SocketAddress remoteAddress) { this.deviceId = deviceId; + this.uniqueId = uniqueId; + this.protocol = protocol; + this.channel = channel; + this.remoteAddress = remoteAddress; } public long getDeviceId() { return deviceId; } + public String getUniqueId() { + return uniqueId; + } + + public Channel getChannel() { + return channel; + } + + public SocketAddress getRemoteAddress() { + return remoteAddress; + } + + public boolean supportsLiveCommands() { + return BasePipelineFactory.getHandler(channel.pipeline(), HttpRequestDecoder.class) == null; + } + + public void sendCommand(Command command) { + protocol.sendDataCommand(channel, remoteAddress, command); + } + public static final String KEY_TIMEZONE = "timezone"; private final Map locals = new HashMap<>(); diff --git a/src/main/java/org/traccar/session/Endpoint.java b/src/main/java/org/traccar/session/Endpoint.java new file mode 100644 index 000000000..76aac3444 --- /dev/null +++ b/src/main/java/org/traccar/session/Endpoint.java @@ -0,0 +1,58 @@ +/* + * Copyright 2022 Anton Tananaev (anton@traccar.org) + * + * 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 org.traccar.session; + +import io.netty.channel.Channel; + +import java.net.SocketAddress; +import java.util.Objects; + +public class Endpoint { + + private final Channel channel; + private final SocketAddress remoteAddress; + + public Endpoint(Channel channel, SocketAddress remoteAddress) { + this.channel = channel; + this.remoteAddress = remoteAddress; + } + + public Channel getChannel() { + return channel; + } + + public SocketAddress getRemoteAddress() { + return remoteAddress; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Endpoint endpoint = (Endpoint) o; + return channel.equals(endpoint.channel) && remoteAddress.equals(endpoint.remoteAddress); + } + + @Override + public int hashCode() { + return Objects.hash(channel, remoteAddress); + } + +} -- cgit v1.2.3