Skip to content

fix(mavlink): harden endpoint transport lifecycle #2164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mbuckton merged 1 commit into from
Aug 10, 2026
Merged

fix(mavlink): harden endpoint transport lifecycle #2164

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion src/main/java/io/mapsmessaging/network/io/impl/udp/UDPFacadeEndPoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class UDPFacadeEndPoint extends EndPoint {
Expand All @@ -40,6 +41,7 @@ public class UDPFacadeEndPoint extends EndPoint {

private final EndPoint endPoint;
private final SocketAddress fromAddress;
private final AtomicBoolean closed = new AtomicBoolean();

public UDPFacadeEndPoint(EndPoint endPoint, SocketAddress fromAddress, EndPointServerStatus server) {
super(counter.incrementAndGet(), server);
Expand Down Expand Up @@ -71,16 +73,25 @@ public String getProtocol() {

@Override
public int sendPacket(Packet packet) throws IOException {
if (closed.get()) {
throw new IOException("UDP facade endpoint is closed");
}
return endPoint.sendPacket(packet);
}

@Override
public int readPacket(Packet packet) throws IOException {
if (closed.get()) {
return -1;
}
return endPoint.readPacket(packet);
}

@Override
public FutureTask<SelectionKey> register(int selectionKey, Selectable runner) throws IOException {
if (closed.get()) {
throw new ClosedChannelException();
}
return endPoint.register(selectionKey, runner);
}

Expand Down Expand Up @@ -112,7 +123,10 @@ public String getRemoteSocketAddress() {

@Override
public void close() throws IOException {
endPoint.close();
if (!closed.compareAndSet(false, true)) {
return;
}
super.close();
EndPointServer endPointServer = (EndPointServer)endPoint.getServer();
endPointServer.handleCloseEndPoint(this);
}
Expand Down
103 changes: 57 additions & 46 deletions src/main/java/io/mapsmessaging/network/protocol/impl/mavlink/MavlinkInterfaceManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

package io.mapsmessaging.network.protocol.impl.mavlink;

import static io.mapsmessaging.logging.ServerLogMessages.MAVLINK_DETECTED_PACKET;
import static io.mapsmessaging.logging.ServerLogMessages.MAVLINK_FAILED_FORWARD_PACKET;
import static io.mapsmessaging.logging.ServerLogMessages.MAVLINK_FAILED_PARSING_FORWARD_LIST;
import static io.mapsmessaging.logging.ServerLogMessages.MAVLINK_FAILED_SETTING_UP_SESSION;
Expand All @@ -30,7 +29,6 @@
import io.mapsmessaging.logging.Logger;
import io.mapsmessaging.logging.LoggerFactory;
import io.mapsmessaging.mavlink.MavlinkEventFactory;
import io.mapsmessaging.mavlink.ProcessedFrame;
import io.mapsmessaging.mavlink.tlog.MavlinkTlogWriter;
import io.mapsmessaging.mavlink.tlog.TlogConfiguration;
import io.mapsmessaging.network.io.EndPoint;
Expand All @@ -39,7 +37,6 @@
import io.mapsmessaging.network.io.impl.SelectorTask;
import io.mapsmessaging.network.io.impl.udp.UDPFacadeEndPoint;
import io.mapsmessaging.network.io.impl.udp.session.UDPSessionState;
import lombok.Getter;

import java.io.IOException;
import java.net.InetSocketAddress;
Expand All @@ -52,7 +49,6 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Optional;

public class MavlinkInterfaceManager implements SelectorCallback, MavlinkConnectionManager {

Expand All @@ -61,7 +57,6 @@ public class MavlinkInterfaceManager implements SelectorCallback, MavlinkConnect
private final SelectorTask selectorTask;
private final EndPoint endPoint;
private final MavLinkSessionManager<MavlinkProtocol> currentSessions;
private final MavlinkEventFactory mavlinkEventFactory;
private final MavlinkConfig mavlinkConfig;
private final List<InetSocketAddress> forwardList;
private final MavlinkTlogWriter tlogWriter;
Expand All @@ -70,7 +65,6 @@ public MavlinkInterfaceManager(EndPoint endPoint) throws IOException {
this.endPoint = endPoint;
mavlinkConfig = (MavlinkConfig) endPoint.getConfig().getProtocolConfig("mavlink");
long timeout = mavlinkConfig.getIdleSessionTimeout();
mavlinkEventFactory = loadDialect(mavlinkConfig.getDialectName());
currentSessions = new MavLinkSessionManager<>(timeout);
selectorTask = new SelectorTask(this, endPoint.getConfig().getEndPointConfig(), endPoint.isUDP());
selectorTask.register(SelectionKey.OP_READ);
Expand Down Expand Up @@ -103,40 +97,50 @@ public static MavlinkEventFactory loadDialect(String name) throws IOException {

@Override
public boolean processPacket(Packet packet) throws IOException {
if (packet.getFromAddress() == null) {
return true;
}
try {
SocketAddress fromAddress = packet.getFromAddress();
if (fromAddress == null) {
return true;
}

byte[] raw = new byte[packet.available()];
int pos = packet.position();
packet.get(raw);
packet.position(pos);

boolean forwardedSource = fromForward(fromAddress);
List<byte[]> packets = MavlinkFrameExtractor.extractMavlinkFrames(raw);
for(byte[] data:packets) {
writeTlog(data);
int systemId = MavlinkFrameExtractor.getSystemId(data);
if (!isAllowedSystem(systemId)) {
if (!forwardedSource) {
forwardPacket(data);
}
continue;
}

byte[] raw = new byte[packet.available()];
int pos = packet.position();
packet.get(raw);
packet.position(pos);

List<byte[]> packets = MavlinkFrameExtractor.extractMavlinkFrames(raw);
for(byte[] data:packets) {
writeTlog(data);
int systemId = MavlinkFrameExtractor.getSystemId(data);
MavlinkDeviceKey key = buildKey(packet, systemId);
boolean allowed =
mavlinkConfig.getAcceptedSources() == null
|| mavlinkConfig.getAcceptedSources().isEmpty()
|| mavlinkConfig.getAcceptedSources().stream().anyMatch(knownSource -> knownSource.getSystemId() == key.getSystemId());

if (allowed) {
MavlinkDeviceKey key = buildKey(packet, systemId);
UDPSessionState<MavlinkProtocol> state = findOrCreate(key);
if (fromForward(packet)) {
state.getContext().processPacket(packet);
} else if (state.getContext() != null) {
MavlinkProtocol protocol = state.getContext();
protocol.processRawFrame(data, packet.getFromAddress().toString());
if (state == null || state.getContext() == null) {
continue;
}

state.getContext().processRawFrame(data, fromAddress.toString());
if (!forwardedSource) {
forwardPacket(data);
}
} else {
forwardPacket(data);
}
return true;
} finally {
selectorTask.register(SelectionKey.OP_READ);
}
selectorTask.register(SelectionKey.OP_READ);
return true;
}

private boolean isAllowedSystem(int systemId) {
return mavlinkConfig.getAcceptedSources() == null
|| mavlinkConfig.getAcceptedSources().isEmpty()
|| mavlinkConfig.getAcceptedSources().stream().anyMatch(knownSource -> knownSource.getSystemId() == systemId);
}

private MavlinkTlogWriter createTlogWriter() throws IOException {
Expand Down Expand Up @@ -171,22 +175,29 @@ private MavlinkDeviceKey buildKey(Packet packet, int systemId) {

private synchronized UDPSessionState<MavlinkProtocol> findOrCreate(MavlinkDeviceKey key) {
UDPSessionState<MavlinkProtocol> state = currentSessions.getState(key);
if (state == null) {
UDPFacadeEndPoint facade = new UDPFacadeEndPoint(endPoint, key.getRemoteAddress(), endPoint.getServer());
if (state != null) {
return state;
}

UDPFacadeEndPoint facade = new UDPFacadeEndPoint(endPoint, key.getRemoteAddress(), endPoint.getServer());
try {
MavlinkProtocol protocol = new MavlinkProtocol(this, key, facade, this.mavlinkConfig);
state = new UDPSessionState<>(protocol);
currentSessions.addState(key, state);
logger.log(MAVLINK_SESSION_CREATED, key.toString());
return state;
} catch (IOException | RuntimeException e) {
try {
MavlinkProtocol protocol = new MavlinkProtocol(this, key, facade, this.mavlinkConfig);
state = new UDPSessionState<>(protocol);
currentSessions.addState(key, state);
logger.log(MAVLINK_SESSION_CREATED, key.toString());
} catch (IOException e) {
logger.log(MAVLINK_FAILED_SETTING_UP_SESSION, key.toString(), e);
facade.close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
logger.log(MAVLINK_FAILED_SETTING_UP_SESSION, key.toString(), e);
return null;
}
return state;
}

private boolean fromForward(Packet packet) {
SocketAddress fromAddress = packet.getFromAddress();
private boolean fromForward(SocketAddress fromAddress) {
return forwardList.stream().anyMatch(forwardAddress -> forwardAddress.equals(fromAddress));
}

Expand Down Expand Up @@ -276,4 +287,4 @@ private static String toSafeFileName(String value) {

return safe.substring(start, end);
}
}
}
69 changes: 39 additions & 30 deletions src/main/java/io/mapsmessaging/network/protocol/impl/mavlink/MavlinkProtocol.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
import io.mapsmessaging.schemas.formatters.MessageFormatterFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
Expand All @@ -81,9 +82,9 @@ public class MavlinkProtocol extends Protocol {
protected final MavlinkConfigDTO mavlinkConfig;
protected Session session;
private final Map<Integer, MavlinkAcceptedSourceDTO> acceptedComponents;
private final SequenceTracker tracker;
private final Map<Integer, SequenceTracker> sequenceTrackers;
private final String outboundTopicName;
protected MavlinkEventFactory mavlinkEventFactory;
protected volatile MavlinkEventFactory mavlinkEventFactory;
protected final MessageFormatter formatter;
private final QualityOfService qos;
private final boolean storeOffline;
Expand All @@ -102,7 +103,7 @@ protected MavlinkProtocol(
super(endPoint, protocolConfig);
this.factory = factory;
this.key = key;
tracker = new SequenceTracker();
sequenceTrackers = new ConcurrentHashMap<>();
this.mavlinkConfig = (MavlinkConfigDTO) protocolConfig;
String dialectName = mavlinkConfig.getDialectName();
mavlinkEventFactory = MavlinkInterfaceManager.loadDialect(dialectName);
Expand Down Expand Up @@ -145,23 +146,30 @@ protected MavlinkProtocol(
session.addSubscription(subscriptionContextBuilder.build());
}
outboundTopicName = outboundTopic;
String remoteSocket = endPoint.getRemoteSocketAddress();
heartbeatEmitter = new MavlinkHeartbeatEmitter(sequenceCounter, endPoint, mavlinkConfig, parseSocketAddress(remoteSocket));
if (mavlinkConfig.hasLocalMavlinkIdentity()) {
SocketAddress heartbeatAddress = endPoint.isUDP() ? parseSocketAddress(endPoint.getRemoteSocketAddress()) : null;
heartbeatEmitter = new MavlinkHeartbeatEmitter(sequenceCounter, endPoint, mavlinkConfig, heartbeatAddress);
} else {
heartbeatEmitter = null;
}
startHeartbeatIfConfigured();
}

@Override
public void close() throws IOException {
stopHeartbeat();
if (!session.isClosed()) {
SessionManager.getInstance().close(session, false);
}
endPoint.close();
if (mbean != null) {
mbean.close();
try {
if (!session.isClosed()) {
SessionManager.getInstance().close(session, false);
}
endPoint.close();
if (mbean != null) {
mbean.close();
}
super.close();
} finally {
factory.close(key);
}
super.close();
factory.close(key);
}

@Override
Expand Down Expand Up @@ -205,14 +213,10 @@ public void sendMessage(@NotNull @NonNull MessageEvent messageEvent) {
}

String json = new String(messageEvent.getMessage().getOpaqueData(), StandardCharsets.UTF_8);
JsonObject input = JsonParser.parseString(json).getAsJsonObject();
String socketAddressText = parts[2];
sendData(input, socketAddressText);
}
catch(Throwable th){
th.printStackTrace();
}
finally {
sendData(JsonParser.parseString(json).getAsJsonObject(), parts[2]);
} catch (RuntimeException e) {
logger.log(MAVLINK_FAILED_SENDING_OUTBOUND_PACKET, endPoint.getName(), "message-event", e);
} finally {
messageEvent.getCompletionTask().run();
}
}
Expand All @@ -223,10 +227,12 @@ private void sendData(JsonObject input, String socketAddressText) {
validateOutboundHeader(input);
byte[] frame = formatter.parseFromJson(input);
Packet packet = new Packet(ByteBuffer.wrap(frame));
packet.setFromAddress(parseSocketAddress(socketAddressText));
if (endPoint.isUDP()) {
packet.setFromAddress(parseSocketAddress(socketAddressText));
}
endPoint.sendPacket(packet);
factory.writeTlog(frame);
} catch (Throwable e) {
} catch (Exception e) {
logger.log(MAVLINK_FAILED_SENDING_OUTBOUND_PACKET, endPoint.getName(), socketAddressText, e);
}
}
Expand Down Expand Up @@ -277,6 +283,8 @@ public void processRawFrame(byte[] raw, String socketAddress) throws IOException
}

if (mavlinkConfig.getStatusTopicNameTemplate() != null && !mavlinkConfig.getStatusTopicNameTemplate().isEmpty()) {
int trackerKey = (env.getFrame().getSystemId() << 8) | env.getFrame().getComponentId();
SequenceTracker tracker = sequenceTrackers.computeIfAbsent(trackerKey, ignored -> new SequenceTracker());
SequenceResult results = tracker.accept(env.getFrame().getSequence());
if (results.isStatusChanged()) {
String statusTopic = computeTopicName(mavlinkConfig.getStatusTopicNameTemplate(), env.getFrame(), env.getMessageName());
Expand All @@ -297,7 +305,7 @@ public void processRawFrame(byte[] raw, String socketAddress) throws IOException
if (env.getDetections() != null && !env.getDetections().isEmpty()) {
envelope.add("detections", gson.toJsonTree(env.getDetections()).getAsJsonArray());
}
raw = envelope.toString().getBytes();
raw = envelope.toString().getBytes(StandardCharsets.UTF_8);
}
processPacket(env.getFrame(), env.getMessageName(), raw, socketAddress);
} else {
Expand Down Expand Up @@ -460,18 +468,19 @@ private static InetSocketAddress parseSocketAddress(String socketAddressText) {


private void startHeartbeatIfConfigured() {
if (!mavlinkConfig.hasLocalMavlinkIdentity()) {
if (heartbeatEmitter == null || heartbeatFuture != null) {
return;
}

long intervalSeconds = Math.max(1, mavlinkConfig.getHeartbeatIntervalSeconds());
SimpleTaskScheduler.getInstance().scheduleAtFixedRate(heartbeatEmitter, intervalSeconds, intervalSeconds, TimeUnit.SECONDS);
heartbeatFuture = SimpleTaskScheduler.getInstance().scheduleAtFixedRate(heartbeatEmitter, intervalSeconds, intervalSeconds, TimeUnit.SECONDS);
}

private void stopHeartbeat() {
if (heartbeatFuture != null) {
heartbeatFuture.cancel(false);
heartbeatFuture = null;
ScheduledFuture<?> future = heartbeatFuture;
heartbeatFuture = null;
if (future != null) {
future.cancel(false);
}
}

Expand Down Expand Up @@ -518,4 +527,4 @@ private int getRequiredUnsignedByte(JsonObject object, String fieldName) {

return value;
}
}
}
Loading
Loading