Add full featured transport based on junixsocket library.

dbus-java-tests/pom.xml

@@ -46,6 +46,7 @@
                             <classpathDependencyExcludes>
                                 <classpathDependencyExclude>${project.groupId}:dbus-java-transport-tcp</classpathDependencyExclude>
                                 <classpathDependencyExclude>${project.groupId}:dbus-java-transport-jnr-unixsocket</classpathDependencyExclude>
+                                <classpathDependencyExclude>${project.groupId}:dbus-java-transport-junixsocket</classpathDependencyExclude>
                             </classpathDependencyExcludes>
                         </configuration>
                     </execution>
@@ -66,6 +67,22 @@
                             <classpathDependencyExcludes>
                                 <classpathDependencyExclude>${project.groupId}:dbus-java-transport-tcp</classpathDependencyExclude>
                                 <classpathDependencyExclude>${project.groupId}:dbus-java-transport-native-unixsocket</classpathDependencyExclude>
+                                <classpathDependencyExclude>${project.groupId}:dbus-java-transport-junixsocket</classpathDependencyExclude>
+                            </classpathDependencyExcludes>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>junixsocket-tests</id>
+                        <phase>test</phase>
+                        <goals>
+                            <goal>test</goal>
+                        </goals>
+                        <configuration>
+                            <skipAfterFailureCount>1</skipAfterFailureCount>
+                            <classpathDependencyExcludes>
+                                <classpathDependencyExclude>${project.groupId}:dbus-java-transport-tcp</classpathDependencyExclude>
+                                <classpathDependencyExclude>${project.groupId}:dbus-java-transport-jnr-unixsocket</classpathDependencyExclude>
+                                <classpathDependencyExclude>${project.groupId}:dbus-java-transport-native-unixsocket</classpathDependencyExclude>
                             </classpathDependencyExcludes>
                         </configuration>
                     </execution>
@@ -80,6 +97,7 @@
                             <classpathDependencyExcludes>
                                 <classpathDependencyExclude>${project.groupId}:dbus-java-transport-native-unixsocket</classpathDependencyExclude>
                                 <classpathDependencyExclude>${project.groupId}:dbus-java-transport-jnr-unixsocket</classpathDependencyExclude>
+                                <classpathDependencyExclude>${project.groupId}:dbus-java-transport-junixsocket</classpathDependencyExclude>
                             </classpathDependencyExcludes>
                         </configuration>
                     </execution>
@@ -109,6 +127,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.github.hypfvieh</groupId>
+            <artifactId>dbus-java-transport-junixsocket</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-core</artifactId>

dbus-java-transport-junixsocket/.gitignore

@@ -0,0 +1 @@
+/target/

dbus-java-transport-junixsocket/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+    <packaging>jar</packaging>
+
+    <artifactId>dbus-java-transport-junixsocket</artifactId>
+
+    <name>${project.artifactId}</name>
+
+    <description>
+        dbus-java transport providing unixsocket support using junixsocket.
+    </description>
+
+    <parent>
+        <groupId>com.github.hypfvieh</groupId>
+        <artifactId>dbus-java-parent</artifactId>
+        <version>4.3.1-SNAPSHOT</version>
+    </parent>
+
+    <properties>
+        <parentDir>${project.parent.basedir}</parentDir>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.github.hypfvieh</groupId>
+            <artifactId>dbus-java-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.kohlschutter.junixsocket</groupId>
+            <artifactId>junixsocket-core</artifactId>
+            <type>pom</type>
+        </dependency>
+
+        <dependency>
+            <groupId>com.kohlschutter</groupId>
+            <artifactId>compiler-annotations</artifactId>
+            <version>1.5.6</version>
+            <scope>compile</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>

dbus-java-transport-junixsocket/src/main/java/module-info.java

@@ -0,0 +1,15 @@
+module org.freedesktop.dbus.transport.junixsocket {
+    requires transitive org.newsclub.net.unix;
+
+    requires transitive org.slf4j;
+
+    requires org.freedesktop.dbus;
+
+    provides org.freedesktop.dbus.spi.transport.ITransportProvider
+            with
+                    org.freedesktop.dbus.transport.junixsocket.JUnixSocketTransportProvider;
+
+    provides org.freedesktop.dbus.spi.message.ISocketProvider
+            with
+                    org.freedesktop.dbus.transport.junixsocket.JUnixSocketSocketProvider;
+}

dbus-java-transport-junixsocket/src/main/java/org/freedesktop/dbus/transport/junixsocket/JUnixSocketMessageReader.java

@@ -0,0 +1,223 @@
+package org.freedesktop.dbus.transport.junixsocket;
+
+import org.freedesktop.dbus.exceptions.DBusException;
+import org.freedesktop.dbus.exceptions.MessageProtocolVersionException;
+import org.freedesktop.dbus.messages.Message;
+import org.freedesktop.dbus.messages.MessageFactory;
+import org.freedesktop.dbus.spi.message.IMessageReader;
+import org.newsclub.net.unix.AFUNIXSocketChannel;
+import org.newsclub.net.unix.FileDescriptorCast;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.EOFException;
+import java.io.FileDescriptor;
+import java.io.IOException;
+import java.net.SocketTimeoutException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class JUnixSocketMessageReader implements IMessageReader {
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    private final AFUNIXSocketChannel socket;
+    private final boolean hasFileDescriptorSupport;
+    private final int[] len = new int[4];
+    private final ByteBuffer buf = ByteBuffer.allocateDirect(12);
+    private final ByteBuffer tbuf = ByteBuffer.allocateDirect(4);
+    private ByteBuffer header = null;
+    private ByteBuffer body = null;
+
+    public JUnixSocketMessageReader(AFUNIXSocketChannel _socket, boolean _hasFileDescriptorSupport) {
+        socket = _socket;
+        hasFileDescriptorSupport = _hasFileDescriptorSupport;
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (socket.isOpen()) {
+            logger.debug("Closing Message Writer");
+            socket.close();
+        }
+    }
+
+    @Override
+    public boolean isClosed() {
+        return !socket.isOpen();
+    }
+
+    @Override
+    public Message readMessage() throws IOException, DBusException {
+        /* Read the 12 byte fixed header, retrying as necessary */
+        if (len[0] < 12) {
+            try {
+                int rv = socket.read(buf);
+                if (rv < 0) {
+                    throw new EOFException("(1) Underlying transport returned " + rv);
+                }
+                len[0] += rv;
+            } catch (SocketTimeoutException _ex) {
+                return null;
+            }
+        }
+
+        if (len[0] == 0) {
+            return null;
+        }
+
+        if (len[0] < 12) {
+            logger.trace("Only got {} of 12 bytes of header", len[0]);
+            return null;
+        }
+
+        buf.flip();
+
+        /* Ensure protocol version. */
+        int protoVer = buf.get(3);
+
+        if (protoVer > Message.PROTOCOL) {
+            throw new MessageProtocolVersionException(String.format("Protocol version %s is unsupported", protoVer));
+        }
+
+        final ByteOrder byteOrder;
+        switch (buf.get(0)) {
+            case Message.Endian.BIG:
+                byteOrder = ByteOrder.BIG_ENDIAN;
+                break;
+            case Message.Endian.LITTLE:
+                byteOrder = ByteOrder.LITTLE_ENDIAN;
+                break;
+            default:
+                throw new IOException(String.format("Unsupported endian: %s", (char) buf.get(0)));
+        }
+
+        buf.order(byteOrder);
+        tbuf.order(byteOrder);
+
+        if (len[1] < 4) {
+            try {
+                int rv = socket.read(tbuf);
+                if (rv < 0) {
+                    throw new EOFException("(2) Underlying transport returned " + rv);
+                }
+                len[1] += rv;
+            } catch (SocketTimeoutException _ex) {
+                return null;
+            }
+        }
+
+        if (len[1] < 4) {
+            logger.trace("Only got {} of 4 bytes of header", len[1]);
+            return null;
+        }
+
+        tbuf.flip();
+
+        /* Parse the variable header length */
+        int headerlen;
+
+        if (header == null) {
+            headerlen = tbuf.getInt(0);
+
+            /* n % 2^i = n & (2^i - 1) */
+            int modlen = headerlen & 7;
+            if (modlen != 0) {
+                headerlen += 8 - modlen;
+            }
+
+            header = ByteBuffer.allocateDirect(headerlen + 8).put(tbuf).position(8);
+            len[2] = 0;
+        } else {
+            headerlen = header.capacity() - 8;
+        }
+
+        if (len[2] < headerlen) {
+            try {
+                int rv = socket.read(header);
+                if (rv < 0) {
+                    throw new EOFException("(3) Underlying transport returned " + rv);
+                }
+                len[2] += rv;
+            } catch (SocketTimeoutException _ex) {
+                return null;
+            }
+        }
+
+        if (len[2] < headerlen) {
+            logger.trace("Only got {} of {} bytes of header", len[2], headerlen);
+            return null;
+        }
+
+        header.flip();
+
+        byte type = buf.get(1);
+
+        /* Read the body */
+        if (body == null) {
+            body = ByteBuffer.allocateDirect(buf.getInt(4));
+            len[3] = 0;
+        }
+
+        if (len[3] < body.capacity()) {
+            try {
+                int rv = socket.read(body);
+                if (rv < 0) {
+                    throw new EOFException("(4) Underlying transport returned " + rv);
+                }
+                len[3] += rv;
+            } catch (SocketTimeoutException _ex) {
+                return null;
+            }
+        }
+
+        if (len[3] < body.capacity()) {
+            logger.trace("Only got {} of {} bytes of body", len[3], body.capacity());
+            return null;
+        }
+
+        body.flip();
+
+        try {
+            final List<org.freedesktop.dbus.FileDescriptor> fds;
+            if (hasFileDescriptorSupport) {
+                FileDescriptor[] receivedFileDescriptors = socket.getReceivedFileDescriptors();
+                if (receivedFileDescriptors.length == 0) {
+                    fds = null;
+                } else {
+                    fds = new ArrayList<>(receivedFileDescriptors.length);
+                    for (int i = 0; i < receivedFileDescriptors.length; i++) {
+                        fds.set(i, new org.freedesktop.dbus.FileDescriptor(FileDescriptorCast.using(receivedFileDescriptors[i]).as(Integer.class)));
+                    }
+                }
+            } else {
+                fds = null;
+            }
+            if (fds != null) {
+                logger.debug("=> {}", fds);
+            }
+            return MessageFactory.createMessage(type,
+                    remaining(buf),
+                    remaining(header),
+                    remaining(body),
+                    fds);
+        } catch (DBusException | RuntimeException _ex) {
+            logger.warn("Exception while creating message.", _ex);
+            throw _ex;
+        } finally {
+            tbuf.rewind();
+            buf.rewind();
+            Arrays.fill(len, 0);
+            body = null;
+            header = null;
+        }
+    }
+
+    private static byte[] remaining(ByteBuffer _buffer) {
+        byte[] bytes = new byte[_buffer.remaining()];
+        _buffer.get(bytes);
+        return bytes;
+    }
+}