Changeset 114:05d0451ae828 in finroc_plugins_tcp-java

Timestamp:

08.06.2017 03:24:53 (6 years ago)

Author:

Max Reichardt <mreichardt@…>

Branch:

17.03

Phase:

public

Message:

Factors out TCP-independent part of protocol (similar to what was done in C++) and adapts to changes Finroc17 (e.g. management of serialization version)

Files:

• Peer.java (modified) (4 diffs)
• TCPSettings.java (modified) (1 diff)
• internal/PeerInfo.java (modified) (2 diffs)
• internal/RemotePart.java (modified) (6 diffs)
• internal/SerializedTCPCommand.java (deleted)
• internal/TCP.java (modified) (7 diffs)
• internal/TCPConnection.java (modified) (42 diffs)
• internal/TCPPeer.java (modified) (20 diffs)
• internal/TCPPort.java (deleted)

Peer.java

@@ -22 +22 @@
 package org.finroc.plugins.tcp;
 
-import org.finroc.core.datatype.FrameworkElementInfo;
 import org.finroc.core.plugin.ExternalConnection;
+import org.finroc.core.remote.Definitions;
 import org.finroc.plugins.tcp.internal.TCP;
 import org.finroc.plugins.tcp.internal.TCPPeer;
@@ -58 +58 @@
             serverListenAddress = "0.0.0.0";
         }
-        implementation = new TCPPeer(this, peerName, TCP.PeerType.FULL, FrameworkElementInfo.StructureExchange.SHARED_PORTS, networkConnection,
+        implementation = new TCPPeer(this, peerName, TCP.PeerType.FULL, Definitions.StructureExchange.SHARED_PORTS, networkConnection,
                                      preferredServerPort, tryNextPortsIfOccupied, autoConnectToAllPeers, serverListenAddress);
     }
@@ -75 +75 @@
             serverListenAddress = "0.0.0.0";
         }
-        implementation = new TCPPeer(this, peerName, TCP.PeerType.SERVER_ONLY, FrameworkElementInfo.StructureExchange.SHARED_PORTS, "",
+        implementation = new TCPPeer(this, peerName, TCP.PeerType.SERVER_ONLY, Definitions.StructureExchange.SHARED_PORTS, "",
                                      preferredServerPort, tryNextPortsIfOccupied, false, serverListenAddress);
     }
@@ -87 +87 @@
      * @param autoConnectToAllPeers Auto-connect to all peers that become known?
      */
-    public Peer(String peerName, String networkConnection, FrameworkElementInfo.StructureExchange structureExchange, boolean autoConnectToAllPeers) {
+    public Peer(String peerName, String networkConnection, Definitions.StructureExchange structureExchange, boolean autoConnectToAllPeers) {
         super("TCP", networkConnection);
         implementation = new TCPPeer(this, peerName, TCP.PeerType.CLIENT_ONLY, structureExchange, networkConnection, -1, false, autoConnectToAllPeers, "");

TCPSettings.java

@@ -50 +50 @@
     public static final int DEQUEUE_QUEUE_SIZE = 50;
 
-    /** Maximum not acknowledged Packet */
-    public static final int MAX_NOT_ACKNOWLEDGED_PACKETS = 0x1F; // 32 (2^x for fast modulo)
-
-    /** Packets considered when calculating avergage ping time */
-    public static final int AVG_PING_PACKETS = 0x7; // 8 (2^x for fast modulo)
-
     /** Help for debugging: insert checks in data stream => more bandwidth */
     public static final boolean DEBUG_TCP = true;

internal/PeerInfo.java

@@ -57 +57 @@
     public ArrayList<InetAddress> addresses = new ArrayList<InetAddress>();
 
-    /** Are we currently connected with this peer? */
-    public volatile boolean connected;
-
     /** True, if there are ongoing attempts to connect to this peer */
     public volatile boolean connecting;
@@ -88 +85 @@
     }
 
+    /** Whether there is currently a connection to this peer */
+    public boolean isConnected() {
+        return remotePart != null && remotePart.isConnected();
+    }
+
     public String toString() {
         return name != null ? (name + " (" + uuid.toString() + ")") : uuid.toString();

internal/RemotePart.java

@@ -22 +22 @@
 package org.finroc.plugins.tcp.internal;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 
 import org.finroc.core.FrameworkElement;
-import org.finroc.core.FrameworkElementFlags;
-import org.finroc.core.FrameworkElementTags;
-import org.finroc.core.LockOrderLevels;
 import org.finroc.core.RuntimeEnvironment;
-import org.finroc.core.admin.AdminClient;
-import org.finroc.core.admin.AdministrationService;
-import org.finroc.core.datatype.FrameworkElementInfo;
-import org.finroc.core.port.AbstractPort;
-import org.finroc.core.port.PortCreationInfo;
-import org.finroc.core.port.cc.CCPortBase;
-import org.finroc.core.port.cc.CCPortDataManager;
-import org.finroc.core.port.cc.CCPortDataManagerTL;
-import org.finroc.core.port.cc.CCPullRequestHandler;
-import org.finroc.core.port.net.NetPort;
-import org.finroc.core.port.rpc.RPCInterfaceType;
-import org.finroc.core.port.rpc.internal.AbstractCall;
-import org.finroc.core.port.rpc.internal.RPCMessage;
-import org.finroc.core.port.rpc.internal.RPCPort;
-import org.finroc.core.port.rpc.internal.RPCRequest;
-import org.finroc.core.port.rpc.internal.RPCResponse;
-import org.finroc.core.port.rpc.internal.AbstractCall.CallType;
-import org.finroc.core.port.std.PortBase;
-import org.finroc.core.port.std.PortDataManager;
-import org.finroc.core.port.std.PullRequestHandler;
-import org.finroc.core.portdatabase.CCType;
-import org.finroc.core.portdatabase.FinrocTypeInfo;
-import org.finroc.core.remote.ModelNode;
-import org.finroc.core.remote.RemoteFrameworkElement;
-import org.finroc.core.remote.RemotePort;
-import org.finroc.core.remote.RemoteRuntime;
-import org.finroc.core.remote.RemoteTypes;
-import org.finroc.plugins.tcp.internal.TCP.OpCode;
+import org.finroc.core.net.generic_protocol.Connection;
+import org.finroc.core.net.generic_protocol.RemoteRuntime;
+import org.finroc.core.remote.BufferedModelChanges;
+import org.finroc.core.remote.Definitions;
+import org.finroc.core.remote.ModelHandler;
 import org.rrlib.logging.Log;
 import org.rrlib.logging.LogLevel;
 import org.rrlib.serialization.BinaryInputStream;
-import org.rrlib.serialization.MemoryBuffer;
-import org.rrlib.serialization.Serialization;
-import org.rrlib.serialization.compression.Compressible;
-import org.rrlib.serialization.rtti.DataTypeBase;
 
 /**
@@ -75 +41 @@
  * It creates a proxy port for each shared port in the remote runtime.
  */
-public class RemotePart extends FrameworkElement implements PullRequestHandler, CCPullRequestHandler {
+public class RemotePart extends RemoteRuntime {
 
     /** Peer info that this part belongs to */
@@ -83 +49 @@
     final TCPPeer peerImplementation;
 
-    /** Connection to transfer "express" ports data */
-    TCPConnection expressConnection;
-
-    /** Connection to transfer "bulk" ports data */
-    TCPConnection bulkConnection;
-
-    /** Connection to transfer all other "management" data */
-    TCPConnection managementConnection;
-
-    /** Structure information to send to remote part */
-    private FrameworkElementInfo.StructureExchange sendStructureInfo = FrameworkElementInfo.StructureExchange.NONE;
-
-    /** Framework element that contains all global links - possibly NULL */
-    private FrameworkElement globalLinks;
-
-    /** Framework element that contains all server ports - possibly NULL */
-    private FrameworkElement serverPorts;
-
-    /**
-     * Lookup for remote framework elements (currently not ports) - similar to remote CoreRegister
-     * (should only be accessed by reader thread of management connection)
-     */
-    private HashMap<Integer, ProxyPort> remotePortRegister = new HashMap<Integer, ProxyPort>();
-
-    /** Temporary buffer for match checks (only used by bulk reader or connector thread) */
-    private final StringBuilder tmpMatchBuffer = new StringBuilder();
-
-    /** Administration interface client port */
-    private final AdminClient adminInterface;
-
-    /** Remote part's current model node */
-    private RemoteRuntime currentModelNode;
-
-    /**
-     * Remote part's new model node. Is created on new connection.
-     * CurrentModelNode is replaced with this, as soon as connection is fully initialized.
-     * As this is not published yet, we can operate on this with any thread.
-     */
-    private RemoteRuntime newModelNode;
-
-    /** List with remote ports that have not been initialized yet */
-    private ArrayList<ProxyPort> uninitializedRemotePorts = new ArrayList<ProxyPort>();
-
-    /** Next call id to assign to sent call */
-    AtomicLong nextCallId = new AtomicLong();
-
-    /** Pull calls that wait for a response */
-    private ArrayList<PullCall> pullCallsAwaitingResponse = new ArrayList<PullCall>();
-
     /** Number of times disconnect was called, since last connect */
     private final AtomicInteger disconnectCalls = new AtomicInteger(0);
 
-    /** Has remote part compression support? */
-    private boolean hasCompressionSupport = false;
-
 
     /** Peer info this part is associated with */
-    RemotePart(PeerInfo peerInfo, FrameworkElement parent, TCPPeer peerImplementation) {
-        super(parent, peerInfo.toString());
+    RemotePart(PeerInfo peerInfo, FrameworkElement parent, TCPPeer peerImplementation, int partnerHandleStampWidth) {
+        super(parent, peerInfo.toString(), peerImplementation.structureExchange == Definitions.StructureExchange.FINSTRUCT, partnerHandleStampWidth);
         this.peerInfo = peerInfo;
         this.peerImplementation = peerImplementation;
-        adminInterface = (peerImplementation.structureExchange == FrameworkElementInfo.StructureExchange.FINSTRUCT) ?
-                         new AdminClient("AdminClient " + getName(), peerImplementation.connectionElement) : null;
     }
 
-    /**
-     * Add connection for this remote part
-     *
-     * @param connection Connection to add (with flags set)
-     * @return Did this succeed? (fails if there already is a connection for specified types of data; may happen if two parts try to connect at the same time - only one connection is kept)
-     */
-    boolean addConnection(TCPConnection connection) {
-        if (connection.bulk) {
-            if (bulkConnection != null) {
-                return false;
-            }
-            bulkConnection = connection;
-        } else {
-            if (expressConnection != null || managementConnection != null) {
-                return false;
-            }
-            expressConnection = connection;
-            managementConnection = connection;
-        }
-
-        if (bulkConnection != null && expressConnection != null && managementConnection != null) {
-            peerInfo.connected = true;
-            Log.log(LogLevel.DEBUG, this, "Connected to " + peerInfo.toString());
-        }
-
-        return true;
+    /** Whether there is currently a connection to this peer */
+    public boolean isConnected() {
+        return getPrimaryConnection() != null && (!((TCPConnection)getPrimaryConnection()).isDisconnected());
     }
 
-    /**
-     * Called during initial structure exchange
-     *
-     * @param info Info on another remote framework element
-     * @param initalStructure Is this call originating from initial structure exchange?
-     * @param remoteRuntime Remote runtime object to add structure to
-     */
-    void addRemoteStructure(FrameworkElementInfo info, boolean initalStructureExchange) {
-        RemoteRuntime remoteRuntime = initalStructureExchange ? newModelNode : currentModelNode;
-        long startTime = 0;
-        while (remoteRuntime == null && (!initalStructureExchange)) {
-            remoteRuntime = currentModelNode;
-            Log.log(LogLevel.DEBUG, this, "Waiting for remote model to become ready");
-            if (startTime == 0) {
-                startTime = System.currentTimeMillis();
-            } else if (System.currentTimeMillis() - startTime > 1000) { // We should not block thread forever (as this blocks GUI interaction in e.g. finstruct)
-                throw new RuntimeException("No model node available");
-            }
-            try {
-                Thread.sleep(20);
-            } catch (InterruptedException e) {}
-        }
-        Log.log(LogLevel.DEBUG_VERBOSE_1, this, "Adding element: " + info.toString());
-        if (info.isPort()) {
-            if ((!hasCompressionSupport) && info.getDataType().getName().equals("finroc.data_compression.CompressionRules")) {
-                hasCompressionSupport = true;
-            }
-            ProxyPort port = new ProxyPort(info);
-            for (int i = 0; i < info.getLinkCount(); i++) {
-                RemoteFrameworkElement remoteElement = new RemotePort(info.getHandle(), info.getLink(i).name, port.getPort(), i);
-                if (i == 0) {
-                    remoteRuntime.elementLookup.put(info.getHandle(), remoteElement);
-                }
-                remoteElement.setName(info.getLink(i).name);
-                remoteElement.setTags(info.getTags());
-                remoteElement.setFlags(info.getFlags());
-                ModelNode parent = getFrameworkElement(info.getLink(i).parent, remoteRuntime);
-                if (initalStructureExchange) {
-                    parent.add(remoteElement);
-                } else {
-                    peerImplementation.connectionElement.getModelHandler().addNode(parent, remoteElement);
-                    if (peerImplementation.structureExchange == FrameworkElementInfo.StructureExchange.SHARED_PORTS) {
-                        port.getPort().init();
-                    } else {
-                        uninitializedRemotePorts.add(port);
-                    }
-                }
-            }
-        } else {
-            RemoteFrameworkElement remoteElement = getFrameworkElement(info.getHandle(), remoteRuntime);
-            remoteElement.setName(info.getLink(0).name);
-            remoteElement.setTags(info.getTags());
-            remoteElement.setFlags(info.getFlags());
-            ModelNode parent = getFrameworkElement(info.getLink(0).parent, remoteRuntime);
-            if (initalStructureExchange) {
-                parent.add(remoteElement);
-            } else {
-                peerImplementation.connectionElement.getModelHandler().addNode(parent, remoteElement);
-            }
-        }
-    }
-
-    /**
-     * Creates new model of remote part
-     */
-    void createNewModel() {
-        newModelNode = new RemoteRuntime(peerInfo.toString(), peerInfo.uuid.toString(), adminInterface, managementConnection.remoteTypes);
+    @Override
+    public void createNewModel() {
+        newModelNode = new org.finroc.core.remote.RemoteRuntime(peerInfo.toString(), peerInfo.uuid.toString(), getAdminInterface(), getPrimaryConnection().getReadBufferStream(), handleStampWidth);
         newModelNode.setFlags(RuntimeEnvironment.getInstance().getAllFlags());
     }
 
-    /**
-     * Creates qualified link for element of remote framework element model
-     *
-     * @param remoteElement Element to create link for
-     * @return Created Link
-     */
-    private String createPortName(RemoteFrameworkElement remoteElement) {
-        return remoteElement.getQualifiedLink();
-    }
-
-    /**
-     * Deletes all child elements of remote part
-     */
-    void deleteAllChildren() {
-        ChildIterator ci = new ChildIterator(this);
-        FrameworkElement child;
-        while ((child = ci.next()) != null) {
-            child.managedDelete();
-        }
-        remotePortRegister = new HashMap<Integer, ProxyPort>();
-    }
-
-    /**
-     * Disconnects remote part
-     */
+    @Override
     public void disconnect() {
         // make sure that disconnect is only called once... prevents deadlocks cleaning up all the threads
@@ -279 +80 @@
 
         synchronized (peerImplementation.connectTo) {
-            if (managementConnection != null) {
-                managementConnection.disconnect();
+            if (getPrimaryConnection() != null) {
+                getPrimaryConnection().disconnect();
             }
-            if (expressConnection != null) {
-                expressConnection.disconnect();
-            }
-            if (bulkConnection != null) {
-                bulkConnection.disconnect();
+            if (getExpressConnection() != null) {
+                getExpressConnection().disconnect();
             }
             disconnectCalls.set(0);
@@ -292 +90 @@
     }
 
-    /**
-     * @param dataRate Data Rate
-     * @return Formatted Data Rate
-     */
-    private static String formatRate(int dataRate) {
-        if (dataRate < 1000) {
-            return "" + dataRate;
-        } else if (dataRate < 10000000) {
-            return (dataRate / 1000) + "k";
-        } else {
-            return (dataRate / 1000000) + "M";
-        }
-    }
-
-    /**
-     * @return Connection quality (see ExternalConnection)
-     */
-    public float getConnectionQuality() {
-        if (bulkConnection == null || expressConnection == null || managementConnection == null || (!peerInfo.connected)) {
-            return 0;
-        }
-        float pingTime = 0;
-        for (int i = 0; i < 3; i++) {
-            TCPConnection c = (i == 0) ? managementConnection : (i == 1 ? bulkConnection : expressConnection);
-            if (c != null) {
-                if (c.pingTimeExceeed()) {
-                    return 0;
-                }
-                pingTime = Math.max(pingTime, (float)c.getAvgPingTime());
-            }
-        }
-        if (pingTime < 300) {
-            return 1;
-        } else if (pingTime > 1300) {
-            return 0;
-        } else {
-            return ((float)pingTime - 300.0f) / 1000.0f;
-        }
-    }
-
-    /**
-     * Returns framework element with specified handle.
-     * Creates one if it doesn't exist.
-     *
-     * @param handle Remote handle of framework element
-     * @param remoteRuntime Remote runtime model to use for lookup
-     * @return Framework element.
-     */
-    public RemoteFrameworkElement getFrameworkElement(int handle, RemoteRuntime remoteRuntime) {
-        RemoteFrameworkElement remoteElement = remoteRuntime.elementLookup.get(handle);
-        if (remoteElement == null) {
-            remoteElement = new RemoteFrameworkElement(handle, "(unknown)");
-            remoteRuntime.elementLookup.put(handle, remoteElement);
-        }
-        return remoteElement;
-    }
-
-    /**
-     * @return Framework element that contains all global links (possibly created by call to this)
-     */
-    public FrameworkElement getGlobalLinkElement() {
-        if (globalLinks == null) {
-            globalLinks = new FrameworkElement(this, "global", Flag.NETWORK_ELEMENT | Flag.GLOBALLY_UNIQUE_LINK | Flag.ALTERNATIVE_LINK_ROOT, -1);
-        }
-        return globalLinks;
-    }
-
-    /**
-     * @return String containing ping times
-     */
-    public String getPingString() {
-        if (!peerInfo.connected) {
-            return "disconnected";
-        }
-
-        int pingAvg = 0;
-        int pingMax = 0;
-        int dataRate = 0;
-        String s = "ping (avg/max/Rx): ";
-        if (bulkConnection == null || expressConnection == null) { // should be disconnected... but maybe this is even safer
-            return s + "- ";
-        }
-        for (int i = 0; i < 3; i++) {
-            TCPConnection c = (i == 0) ? managementConnection : (i == 1 ? bulkConnection : expressConnection);
-            if (c != null) {
-                if (c.pingTimeExceeed()) {
-                    return s + "- ";
-                }
-                pingAvg = Math.max(pingAvg, c.getAvgPingTime());
-                pingMax = Math.max(pingMax, c.getMaxPingTime());
-                dataRate += c.getRx();
-            }
-        }
-        return s + pingAvg + "ms/" + pingMax + "ms/" + formatRate(dataRate);
-    }
-
-    /**
-     * Initializes part and checks for admin port to connect to
-     *
-     * @param obsoleteNode Model node that is now obsolete
-     */
-    public void initAndCheckForAdminPort(final ModelNode obsoleteNode) {
-
-        if (peerImplementation.structureExchange != FrameworkElementInfo.StructureExchange.SHARED_PORTS) {
-
-            // expand port names
-            synchronized (getRegistryLock()) {
-                for (ProxyPort pp : remotePortRegister.values()) {
-                    RemotePort[] remotePorts = RemotePort.get(pp.getPort());
-                    for (int i = 0; i < remotePorts.length; i++) {
-                        pp.getPort().setName(createPortName(remotePorts[i]), i);
-                    }
-                }
-            }
-        }
-
-        this.init();
-
-        // connect to admin interface?
-        if (adminInterface != null) {
-            FrameworkElement fe = getChildElement(AdministrationService.QUALIFIED_PORT_NAME, false);
-            if (fe != null && fe.isPort() && fe.isReady()) {
-                ((AbstractPort)fe).connectTo(adminInterface.getWrapped());
-            } else {
-                Log.log(LogLevel.ERROR, this, "Could not find administration port to connect to.");
-            }
-        }
-
-        // set remote type in RemoteRuntime Annotation
-        //((RemoteRuntime)getAnnotation(RemoteRuntime.class)).setRemoteTypes(managementConnection.updateTimes);
-        final RemoteRuntime oldModel = currentModelNode;
-        final RemoteRuntime newModel = newModelNode;
-        currentModelNode = newModelNode;
-        newModelNode = null;
-
-        if (oldModel != null) {
-            peerImplementation.connectionElement.getModelHandler().removeNode(obsoleteNode);
-            peerImplementation.connectionElement.getModelHandler().replaceNode(oldModel, newModel);
-        } else {
-            peerImplementation.connectionElement.getModelHandler().replaceNode(obsoleteNode, newModel);
-        }
-    }
-
-    /**
-     * Process message with specified opcode in provided stream
-     *
-     * @param opCode Opcode of message
-     * @param stream Stream to read message from
-     * @param remoteTypes Info on remote types
-     * @param connection Connection this was called from
-     * @param elementInfoBuffer Framework element info buffer to use
-     */
-    public void processMessage(OpCode opCode, BinaryInputStream stream, RemoteTypes remoteTypes, TCPConnection connection) throws Exception {
-        Log.log(LogLevel.DEBUG_VERBOSE_1, this, "Processing message " + opCode.toString());
-
-        switch (opCode) {
-        case PORT_VALUE_CHANGE:
-        case SMALL_PORT_VALUE_CHANGE:
-        case SMALL_PORT_VALUE_CHANGE_WITHOUT_TIMESTAMP:
-            int handle = stream.readInt();
-            Serialization.DataEncoding encoding = stream.readEnum(Serialization.DataEncoding.class);
-            AbstractPort port = RuntimeEnvironment.getInstance().getPort(handle);
-            if (port != null && port.isReady() && (!FinrocTypeInfo.isMethodType(port.getDataType(), true))) {
-                NetPort netPort = port.asNetPort();
-                if (netPort != null) {
-                    netPort.receiveDataFromStream(stream, encoding, opCode != OpCode.SMALL_PORT_VALUE_CHANGE_WITHOUT_TIMESTAMP);
-                }
-            }
-            break;
-        case RPC_CALL:
-            handle = stream.readInt();
-            CallType callType = stream.readEnum(CallType.class);
-            DataTypeBase type = stream.readType();
-            byte methodId = stream.readByte();
-            if (!(type instanceof RPCInterfaceType)) {
-                Log.log(LogLevel.WARNING, this, "Type " + type.getName() + " is no RPC type. Ignoring call.");
-                return;
-            }
-            RPCInterfaceType rpcInterfaceType = (RPCInterfaceType)type;
-
-            if (callType == CallType.RPC_MESSAGE || callType == CallType.RPC_REQUEST) {
-                port = RuntimeEnvironment.getInstance().getPort(handle);
-                if (port != null && rpcInterfaceType == port.getDataType()) {
-                    //RPCDeserializationScope deserializationScope(message.Get<0>(), connection.rpcCallBufferPools); // TODO?
-                    if (callType == CallType.RPC_MESSAGE) {
-                        RPCMessage.deserializeAndExecuteCallImplementation(stream, (RPCPort)port, methodId);
-                    } else {
-                        RPCRequest.deserializeAndExecuteCallImplementation(stream, (RPCPort)port, methodId, connection);
-                    }
-                }
-            } else { // type is RPC response
-                long callId = stream.readLong();
-
-                AbstractCall callAwaitingThisResponse = null;
-                synchronized (connection.callsAwaitingResponse) {
-                    for (TCPConnection.CallAndTimeout call : connection.callsAwaitingResponse) {
-                        if (call.call.getCallId() == callId) {
-                            callAwaitingThisResponse = call.call;
-                            connection.callsAwaitingResponse.remove(call);
-                            break;
-                        }
-                    }
-                }
-                if (callAwaitingThisResponse == null) { // look in other connection; TODO: think about rules which connection to use for RPCs
-                    TCPConnection otherConnection = (connection != expressConnection) ? expressConnection : bulkConnection;
-                    synchronized (otherConnection.callsAwaitingResponse) {
-                        for (TCPConnection.CallAndTimeout call : otherConnection.callsAwaitingResponse) {
-                            if (call.call.getCallId() == callId) {
-                                callAwaitingThisResponse = call.call;
-                                otherConnection.callsAwaitingResponse.remove(call);
-                                break;
-                            }
-                        }
-                    }
-                }
-                if (callAwaitingThisResponse != null) {
-                    port = RuntimeEnvironment.getInstance().getPort(callAwaitingThisResponse.getLocalPortHandle());
-                    if (port != null) {
-                        //RPCDeserializationScope deserializationScope(callAwaitingThisResponse.getLocalPortHandle(), connection.rpcCallBufferPools); // TODO?
-                        RPCResponse.deserializeAndExecuteCallImplementation(stream, rpcInterfaceType.getMethod(methodId), connection, callAwaitingThisResponse);
-                        return;
-                    }
-                }
-                RPCResponse.deserializeAndExecuteCallImplementation(stream, rpcInterfaceType.getMethod(methodId), connection, null);
-            }
-            break;
-        case PULLCALL:
-            handle = stream.readInt();
-            long callUid = stream.readLong();
-            encoding = stream.readEnum(Serialization.DataEncoding.class);
-
-            SerializedTCPCommand pullCallReturn = new SerializedTCPCommand(TCP.OpCode.PULLCALL_RETURN, 8192);
-            pullCallReturn.getWriteStream().writeLong(callUid);
-            port = RuntimeEnvironment.getInstance().getPort(handle);
-            if (port != null && port.isReady() && (!FinrocTypeInfo.isMethodType(port.getDataType(), true))) {
-                pullCallReturn.getWriteStream().writeBoolean(false);
-                if (FinrocTypeInfo.isStdType(port.getDataType())) {
-                    CCPortBase pb = (CCPortBase)port;
-                    CCPortDataManager manager = pb.getPullInInterthreadContainerRaw(true, true);
-                    pullCallReturn.getWriteStream().writeBoolean(true);
-                    pullCallReturn.getWriteStream().writeType(manager.getObject().getType());
-                    manager.getTimestamp().serialize(pullCallReturn.getWriteStream());
-                    manager.getObject().serialize(pullCallReturn.getWriteStream(), encoding);
-                    manager.recycle2();
-                } else {
-                    PortBase pb = (PortBase)port;
-                    PortDataManager manager = pb.getPullLockedUnsafe(true, true);
-                    pullCallReturn.getWriteStream().writeBoolean(true);
-                    pullCallReturn.getWriteStream().writeType(manager.getType());
-                    manager.getTimestamp().serialize(pullCallReturn.getWriteStream());
-                    manager.getObject().serialize(pullCallReturn.getWriteStream(), encoding);
-                    manager.releaseLock();
-                }
-            } else {
-                pullCallReturn.getWriteStream().writeBoolean(false);
-            }
-            connection.sendCall(pullCallReturn);
-            break;
-        case PULLCALL_RETURN:
-            callUid = stream.readLong();
-            boolean failed = stream.readBoolean();
-
-            synchronized (pullCallsAwaitingResponse) {
-                PullCall pullCall = null;
-                for (PullCall pullCallWaiting : pullCallsAwaitingResponse) {
-                    if (pullCallWaiting.callId == callUid) {
-                        pullCall = pullCallWaiting;
-                        break;
-                    }
-                }
-                if (pullCall != null) {
-                    synchronized (pullCall) {
-                        if (!failed) {
-                            type = stream.readType();
-                            if (pullCall.origin != null) {
-                                if (pullCall.origin.getDataType() == type) {
-                                    PortDataManager manager = pullCall.origin.getUnusedBufferRaw();
-                                    manager.getTimestamp().deserialize(stream);
-                                    manager.getObject().deserialize(stream, pullCall.encoding);
-                                }
-                            } else if (pullCall.ccResultBuffer != null) {
-                                if (pullCall.ccResultBuffer.getObject().getType() == type) {
-                                    pullCall.ccResultBuffer.getTimestamp().deserialize(stream);
-                                    pullCall.ccResultBuffer.getObject().deserialize(stream, pullCall.encoding);
-                                }
-                            }
-                        }
-                        pullCall.notify();
-                    }
-                }
-            }
-            break;
-        case TYPE_UPDATE:
-            type = stream.readType();
-            remoteTypes.setTime(type, stream.readShort());
-            break;
-
-//        case SUBSCRIBE:
-//            // TODO
-//            SubscribeMessage message;
-//            message.deserialize(stream);
-//
-//            // Get or create server port
-//            auto it = serverPortMap.find(message.Get<0>());
-//            if (it != serverPortMap.end()) {
-//                dataPorts::GenericPort port = it.second;
-//                NetworkPortInfo* networkPortInfo = port.GetAnnotation<NetworkPortInfo>();
-//                networkPortInfo.setServerSideSubscriptionData(message.Get<1>(), message.Get<2>(), message.Get<3>(), message.Get<5>());
-//
-//                boolean pushStrategy = message.Get<1>() > 0;
-//                boolean reversePushStrategy = message.Get<2>();
-//                if (port.getWrapped().pushStrategy() != pushStrategy || port.getWrapped().reversePushStrategy() != reversePushStrategy) {
-//                    // flags need to be changed
-//                    thread::Lock lock(getStructureMutex(), false);
-//                    if (lock.tryLock()) {
-//                        if (port.getWrapped().pushStrategy() != pushStrategy) {
-//                            port.getWrapped().setPushStrategy(pushStrategy);
-//                        }
-//                        if (port.getWrapped().reversePushStrategy() != reversePushStrategy) {
-//                            port.getWrapped().setReversePushStrategy(reversePushStrategy);
-//                        }
-//                    }
-//                    else {
-//                        return true; // We could not obtain lock - try again later
-//                    }
-//                }
-//            }
-//            else {
-//                thread::Lock lock(getStructureMutex(), false);
-//                if (lock.tryLock()) {
-//                    // Create server port
-//                    AbstractPort* port = RuntimeEnvironment::getInstance().getPort(message.Get<0>());
-//                    if ((!port) || (!port.isReady())) {
-//                        fINROC_LOG_PRINT(DEBUG_WARNING, "Port for subscription not available");
-//                        return false;
-//                    }
-//
-//                    Flags flags = Flag::NETWORK_ELEMENT | Flag::VOLATILE;
-//                    if (port.isOutputPort()) {
-//                        flags |= Flag::ACCEPTS_DATA; // create input port
-//                    }
-//                    else {
-//                        flags |= Flag::OUTPUT_PORT | Flag::EMITS_DATA; // create output io port
-//                    }
-//                    if (sendStructureInfo != StructureExchange::SHARED_PORTS) {
-//                        flags |= Flag::TOOL_PORT;
-//                    }
-//                    if (message.Get<1>() > 0) {
-//                        flags |= Flag::PUSH_STRATEGY;
-//                    }
-//                    if (message.Get<2>()) {
-//                        flags |= Flag::PUSH_STRATEGY_REVERSE;
-//                    }
-//
-//                    dataPorts::GenericPort createdPort(port.getQualifiedName().substr(1), getServerPortsElement(), port.getDataType(), flags, message.Get<3>());
-//                    NetworkPortInfo* networkPortInfo = new NetworkPortInfo(*this, message.Get<4>(), message.Get<1>(), true, *createdPort.getWrapped());
-//                    networkPortInfo.setServerSideSubscriptionData(message.Get<1>(), message.Get<2>(), message.Get<3>(), message.Get<5>());
-//                    createdPort.addPortListenerForPointer(*networkPortInfo);
-//                    createdPort.init();
-//                    createdPort.connectTo(port);
-//                    serverPortMap.insert(pair<FrameworkElementHandle, dataPorts::GenericPort>(message.Get<0>(), createdPort));
-//                    fINROC_LOG_PRINT(DEBUG, "Created server port ", createdPort.getWrapped().getQualifiedName());
-//                }
-//                else {
-//                    return true; // We could not obtain lock - try again later
-//                }
-//            }
-//            break;
-//        case UNSUBSCRIBE:
-//            // TODO
-//            UnsubscribeMessage message;
-//            message.deserialize(stream);
-//            auto it = serverPortMap.find(message.Get<0>());
-//            if (it != serverPortMap.end()) {
-//                thread::Lock lock(getStructureMutex(), false);
-//                if (lock.tryLock()) {
-//                    it.second.getWrapped().managedDelete();
-//                    serverPortMap.erase(message.Get<0>());
-//                }
-//                else {
-//                    return true; // We could not obtain lock - try again later
-//                }
-//            }
-//            else {
-//                fINROC_LOG_PRINT(DEBUG_WARNING, "Port for unsubscribing not available");
-//                return false;
-//            }
-//            break;
-
-        case PEER_INFO:
-            PeerInfo info;
+    @Override
+    public void processMessage(TCP.OpCode opCode, BinaryInputStream stream, Connection connection, BufferedModelChanges modelChanges) throws Exception {
+        if (opCode == TCP.OpCode.PEER_INFO) {
             while (stream.readBoolean()) {
                 PeerInfo peer = peerImplementation.deserializePeerInfo(stream);
@@ -688 +98 @@
                 peerImplementation.processIncomingPeerInfo(peer);
             }
-            break;
-        default:
-            throw new Exception("Opcode " + opCode.toString() + " not implemented yet.");
-        }
-    }
-
-    /**
-     * Processes buffer containing change events regarding remote runtime.
-     * Must be executed from model thread.
-     *
-     * @param structureBufferToProcess Buffer containing serialized structure changes
-     * @param remoteTypes Info on remote types
-     */
-    public void processStructurePacket(MemoryBuffer structureBufferToProcess, RemoteTypes remoteTypes) {
-        FrameworkElementInfo info = new FrameworkElementInfo();
-        BinaryInputStream stream = new BinaryInputStream(structureBufferToProcess, remoteTypes);
-        while (stream.moreDataAvailable()) {
-            TCP.OpCode opcode = stream.readEnum(TCP.OpCode.class);
-            if (opcode == TCP.OpCode.STRUCTURE_CREATE) {
-                info.deserialize(stream, peerImplementation.structureExchange);
-                addRemoteStructure(info, false);
-            } else if (opcode == TCP.OpCode.STRUCTURE_CHANGE) {
-                int handle = stream.readInt();
-                int flags = stream.readInt();
-                short strategy = stream.readShort();
-                short updateInterval = stream.readShort();
-                ProxyPort port = remotePortRegister.get(handle);
-                if (port != null) {
-                    boolean connections = peerImplementation.structureExchange == FrameworkElementInfo.StructureExchange.FINSTRUCT;
-                    if (connections) {
-                        info.deserializeConnections(stream);
-                    }
-                    port.update(flags, strategy, updateInterval, connections ? info.copyConnections() : null, connections ? info.copyNetworkConnections() : null);
-
-                    RemotePort[] modelElements = RemotePort.get(port.getPort());
-                    for (RemotePort modelElement : modelElements) {
-                        modelElement.setFlags(flags);
-                    }
-                } else {
-                    RemoteFrameworkElement element = currentModelNode.getRemoteElement(handle);
-                    if (element != null) {
-                        element.setFlags(flags);
-                    }
-                }
-            } else if (opcode == TCP.OpCode.STRUCTURE_DELETE) {
-                int handle = stream.readInt();
-                ProxyPort port = remotePortRegister.get(handle);
-                if (port != null) {
-                    RemotePort[] modelElements = RemotePort.get(port.getPort());
-                    port.managedDelete();
-                    for (RemotePort modelElement : modelElements) {
-                        peerImplementation.connectionElement.getModelHandler().removeNode(modelElement);
-                    }
-                    currentModelNode.elementLookup.remove(handle);
-                    uninitializedRemotePorts.remove(port);
-                } else {
-                    RemoteFrameworkElement element = currentModelNode.getRemoteElement(handle);
-                    if (element != null) {
-                        peerImplementation.connectionElement.getModelHandler().removeNode(element);
-                        currentModelNode.elementLookup.remove(handle);
-                    }
-                }
-            } else {
-                Log.log(LogLevel.WARNING, this, "Received corrupted structure info. Skipping packet");
-                return;
-            }
-            TCPConnection.checkCommandEnd(stream);
-        }
-
-        // Initialize ports whose links are now complete
-        synchronized (getRegistryLock()) {
-            for (int i = 0; i < uninitializedRemotePorts.size(); i++) {
-                ProxyPort port = uninitializedRemotePorts.get(i);
-                RemotePort[] remotePorts = RemotePort.get(port.getPort());
-                boolean complete = true;
-                for (RemotePort remotePort : remotePorts) {
-                    complete |= remotePort.isNodeAncestor(currentModelNode);
-                }
-                if (complete) {
-                    for (int j = 0; j < remotePorts.length; j++) {
-                        port.getPort().setName(createPortName(remotePorts[j]), j);
-                    }
-                    port.getPort().init();
-                    uninitializedRemotePorts.remove(i);
-                    i--;
-                }
-            }
+        } else {
+            super.processMessage(opCode, stream, connection, modelChanges);
         }
     }
 
     @Override
-    public boolean pullRequest(CCPortBase origin, CCPortDataManagerTL resultBuffer, boolean intermediateAssign) {
-        NetPort netport = origin.asNetPort();
-        if (netport != null && expressConnection != null) {
-            PullCall pullCall = new PullCall(netport);
-            pullCall.ccResultBuffer = resultBuffer;
-            synchronized (pullCallsAwaitingResponse) {
-                pullCallsAwaitingResponse.add(pullCall);
-            }
-            synchronized (pullCall) {
-                expressConnection.sendCall(pullCall);
-                try {
-                    pullCall.wait(1000);
-                } catch (InterruptedException e) {}
-            }
-            synchronized (pullCallsAwaitingResponse) {
-                pullCallsAwaitingResponse.remove(pullCall);
-            }
-            if (pullCall.ccPullSuccess) {
-                return true;
-            }
-        }
-        origin.getRaw(resultBuffer.getObject(), true);
-        return true;
+    public void removeConnection(Connection connection) {
+        super.removeConnection(connection);
+        peerInfo.lastConnection = System.currentTimeMillis();
     }
 
     @Override
-    public PortDataManager pullRequest(PortBase origin, byte addLocks, boolean intermediateAssign) {
-        NetPort netport = origin.asNetPort();
-        if (netport != null && expressConnection != null) {
-            PullCall pullCall = new PullCall(netport);
-            pullCall.origin = origin;
-            synchronized (pullCallsAwaitingResponse) {
-                pullCallsAwaitingResponse.add(pullCall);
-            }
-            synchronized (pullCall) {
-                expressConnection.sendCall(pullCall);
-                try {
-                    pullCall.wait(1000);
-                } catch (InterruptedException e) {}
-            }
-            synchronized (pullCallsAwaitingResponse) {
-                pullCallsAwaitingResponse.remove(pullCall);
-            }
-            if (pullCall.resultBuffer != null) {
-                pullCall.resultBuffer.getCurrentRefCounter().setLocks((byte)(addLocks));
-                return pullCall.resultBuffer;
-            }
-        }
-        PortDataManager pd = origin.lockCurrentValueForRead();
-        pd.getCurrentRefCounter().addLocks((byte)(addLocks - 1)); // we already have one lock
-        return pd;
-    }
-
-    /**
-     * Removes connection for this remote part
-     *
-     * @param connection Connection to remove
-     */
-    void removeConnection(TCPConnection connection) {
-        if (connection == managementConnection) {
-            managementConnection = null;
-            deleteAllChildren();
-            globalLinks = null;
-            serverPorts = null;
-            uninitializedRemotePorts.clear();
-            pullCallsAwaitingResponse.clear();
-            if (currentModelNode != null) {
-                peerImplementation.connectionElement.getModelHandler().removeNode(currentModelNode);
-            }
-            currentModelNode = null;
-        }
-        if (connection == expressConnection) {
-            expressConnection = null;
-        }
-        if (connection == bulkConnection) {
-            bulkConnection = null;
-        }
-        if (peerInfo.connected) {
-            peerInfo.lastConnection = System.currentTimeMillis();
-        }
-        peerInfo.connected = false;
-    }
-
-    /**
-     * @param sendStructureInfo Structure information to send to remote part
-     */
-    void setDesiredStructureInfo(FrameworkElementInfo.StructureExchange sendStructureInfo) {
-        if (sendStructureInfo == FrameworkElementInfo.StructureExchange.NONE) {
-            return;
-        }
-        if (this.sendStructureInfo != FrameworkElementInfo.StructureExchange.NONE && this.sendStructureInfo != sendStructureInfo) {
-            Log.log(LogLevel.WARNING, this, "Desired structure info already set to " + this.sendStructureInfo.toString() + ". This is likely to cause trouble.");
-        }
-        this.sendStructureInfo = sendStructureInfo;
-    }
-
-    /**
-     * Local port that acts as proxy for ports on remote machines
-     */
-    public class ProxyPort extends TCPPort {
-
-        /** Has port been found again after reconnect? */
-        private boolean refound = true;
-
-        /** >= 0 when port has subscribed to server; value of current subscription */
-        private short subscriptionStrategy = -1;
-
-        /** true, if current subscription includes reverse push strategy */
-        private boolean subscriptionRevPush = false;
-
-        /** Update time of current subscription */
-        private short subscriptionUpdateTime = -1;
-
-        /** Handles (remote) of port's outgoing connections */
-        protected ArrayList<FrameworkElementInfo.ConnectionInfo> connections = new ArrayList<FrameworkElementInfo.ConnectionInfo>();
-
-        /** Info on port's outgoing network connections */
-        protected ArrayList<FrameworkElementInfo.NetworkConnection> networkConnections = new ArrayList<FrameworkElementInfo.NetworkConnection>();
-
-
-        /**
-         * @param portInfo Port information
-         */
-        public ProxyPort(FrameworkElementInfo portInfo) {
-            super(createPCI(portInfo), null /*(portInfo.getFlags() & Flag.EXPRESS_PORT) > 0 ? expressConnection : bulkConnection // bulkConnection might be null*/);
-            remoteHandle = portInfo.getHandle();
-            remotePortRegister.put(remoteHandle, this);
-
-            super.updateFlags(portInfo.getFlags());
-            getPort().setMinNetUpdateInterval(portInfo.getMinNetUpdateInterval());
-            updateIntervalPartner = portInfo.getMinNetUpdateInterval(); // TODO redundant?
-            propagateStrategyFromTheNet(portInfo.getStrategy());
-            connections = portInfo.copyConnections();
-            networkConnections = portInfo.copyNetworkConnections();
-
-            Log.log(LogLevel.DEBUG_VERBOSE_2, this, "Updating port info: " + portInfo.toString());
-            for (int i = 1, n = portInfo.getLinkCount(); i < n; i++) {
-                FrameworkElement parent = portInfo.getLink(i).unique ? getGlobalLinkElement() : (FrameworkElement)RemotePart.this;
-                getPort().link(parent, portInfo.getLink(i).name);
-            }
-            FrameworkElement parent = portInfo.getLink(0).unique ? getGlobalLinkElement() : (FrameworkElement)RemotePart.this;
-            getPort().setName(portInfo.getLink(0).name);
-            parent.addChild(getPort());
-            FrameworkElementTags.addTags(getPort(), portInfo.getTags());
-
-            if (getPort() instanceof CCPortBase) {
-                ((CCPortBase)getPort()).setPullRequestHandler(RemotePart.this);
-            } else if (getPort() instanceof PortBase) {
-                ((PortBase)getPort()).setPullRequestHandler(RemotePart.this);
-            }
-        }
-
-        public void update(int flags, short strategy, short minNetUpdateInterval, ArrayList<FrameworkElementInfo.ConnectionInfo> newConnections,
-                           ArrayList<FrameworkElementInfo.NetworkConnection> newNetworkConnections) {
-            updateFlags(flags);
-            getPort().setMinNetUpdateInterval(minNetUpdateInterval);
-            updateIntervalPartner = minNetUpdateInterval; // TODO redundant?
-            propagateStrategyFromTheNet(strategy);
-            if (newConnections != null) {
-                connections = newConnections;
-            } else if (connections.size() > 0) {
-                connections = new ArrayList<FrameworkElementInfo.ConnectionInfo>(); // create new list for thread-safety reasons
-            }
-            if (newNetworkConnections != null) {
-                networkConnections = newNetworkConnections;
-            } else if (connections.size() > 0) {
-                networkConnections = new ArrayList<FrameworkElementInfo.NetworkConnection>(); // create new list for thread-safety reasons
-            }
-        }
-
-//        /**
-//         * Is port the one that is described by this information?
-//         *
-//         * @param info Port information
-//         * @return Answer
-//         */
-//        public boolean matches(FrameworkElementInfo info) {
-//            synchronized (getPort()) {
-//                if (remoteHandle != info.getHandle() || info.getLinkCount() != getPort().getLinkCount()) {
-//                    return false;
-//                }
-//                if ((getPort().getAllFlags() & Flag.CONSTANT_FLAGS) != (info.getFlags() & Flag.CONSTANT_FLAGS)) {
-//                    return false;
-//                }
-//                for (int i = 0; i < info.getLinkCount(); i++) {
-//                    if (peerImplementation.structureExchange == FrameworkElementInfo.StructureExchange.SHARED_PORTS) {
-//                        getPort().getQualifiedLink(tmpMatchBuffer, i);
-//                    } else {
-//                        tmpMatchBuffer.delete(0, tmpMatchBuffer.length());
-//                        tmpMatchBuffer.append(getPort().getLink(i).getName());
-//                    }
-//                    if (!tmpMatchBuffer.equals(info.getLink(i).name)) {
-//                        return false;
-//                    }
-//                    // parents are negligible if everything else, matches
-//                }
-//                return true;
-//            }
-//        }
-
-//        public void reset() {
-//            connection = null; // set connection to null
-//            monitored = false; // reset monitored flag
-//            refound = false; // reset refound flag
-//            propagateStrategyFromTheNet((short)0);
-//            subscriptionRevPush = false;
-//            subscriptionUpdateTime = -1;
-//            subscriptionStrategy = -1;
-//        }
-
-//        /**
-//         * Update port properties/information from received port information
-//         *
-//         * @param portInfo Port info
-//         */
-//        private void updateFromPortInfo(FrameworkElementInfo portInfo, TCP.OpCode opCode) {
-//            synchronized (getPort().getRegistryLock()) {
-//                updateFlags(portInfo.getFlags());
-//                getPort().setMinNetUpdateInterval(portInfo.getMinNetUpdateInterval());
-//                updateIntervalPartner = portInfo.getMinNetUpdateInterval(); // TODO redundant?
-//                propagateStrategyFromTheNet(portInfo.getStrategy());
-//                portInfo.getConnections(connections);
-//
-//                log(LogLevel.LL_DEBUG_VERBOSE_2, this, "Updating port info: " + portInfo.toString());
-//                if (opCode == TCP.OpCode.STRUCTURE_CREATE) {
-//                    assert(!getPort().isReady());
-//                    for (int i = 1, n = portInfo.getLinkCount(); i < n; i++) {
-//                        FrameworkElement parent = portInfo.getLink(i).unique ? getGlobalLinkElement() : (FrameworkElement)RemotePart.this;
-//                        getPort().link(parent, portInfo.getLink(i).name);
-//                    }
-//                    FrameworkElement parent = portInfo.getLink(0).unique ? getGlobalLinkElement() : (FrameworkElement)RemotePart.this;
-//                    getPort().setName(portInfo.getLink(0).name);
-//                    parent.addChild(getPort());
-//                }
-//                FrameworkElementTags.addTags(getPort(), portInfo.getTags());
-//
-//                checkSubscription();
-//            }
-//        }
-
-        @Override
-        protected void prepareDelete() {
-            remotePortRegister.remove(remoteHandle);
-            getPort().disconnectAll();
-            checkSubscription();
-            super.prepareDelete();
-        }
-
-        @Override
-        protected void connectionRemoved() {
-            checkSubscription();
-        }
-
-        @Override
-        protected void connectionAdded() {
-            checkSubscription();
-        }
-
-        @Override
-        protected void propagateStrategyOverTheNet() {
-            checkSubscription();
-        }
-
-        @Override
-        protected void checkSubscription() {
-            if (FinrocTypeInfo.isMethodType(getPort().getDataType(), true)) {
-                return;
-            }
-
-            synchronized (getPort().getRegistryLock()) {
-                AbstractPort p = getPort();
-                boolean revPush = p.isInputPort() && (p.isConnectedToReversePushSources() || p.getOutgoingConnectionCount() > 0);
-                short time = getUpdateIntervalForNet();
-                short strategy = p.isInputPort() ? 0 : p.getStrategy();
-                if (!p.isConnected()) {
-                    strategy = -1;
-                }
-
-                TCPConnection c = connection;
-
-                if (c == null) {
-                    subscriptionStrategy = -1;
-                    subscriptionRevPush = false;
-                    subscriptionUpdateTime = -1;
-                } else if (strategy == -1 && subscriptionStrategy > -1) { // disconnect
-                    //System.out.println("Unsubscribing " + (((long)remoteHandle) + (1L << 32L)) + " " + getPort().getQualifiedName());
-                    c.unsubscribe(remoteHandle);
-                    subscriptionStrategy = -1;
-                    subscriptionRevPush = false;
-                    subscriptionUpdateTime = -1;
-                } else if (strategy == -1) {
-                    // still disconnected
-                } else if (strategy != subscriptionStrategy || time != subscriptionUpdateTime || revPush != subscriptionRevPush) {
-                    boolean requestCompressedData = hasCompressionSupport && getNetworkEncoding() == Serialization.DataEncoding.BINARY && (!peerInfo.uuid.hostName.equals(peerImplementation.thisPeer.uuid)) &&
-                                                    isStdType() && getPort().getDataType().getJavaClass() != null && Compressible.class.isAssignableFrom(getPort().getDataType().getJavaClass());
-
-                    c.subscribe(remoteHandle, strategy, revPush, time, p.getHandle(), requestCompressedData ? Serialization.DataEncoding.BINARY_COMPRESSED : getNetworkEncoding());
-                    //c.subscribe(remoteHandle, strategy, revPush, time, p.getHandle(), getNetworkEncoding());
-                    subscriptionStrategy = strategy;
-                    subscriptionRevPush = revPush;
-                    subscriptionUpdateTime = time;
-                }
-                setMonitored(publishPortDataOverTheNet() && getPort().isConnected());
-            }
-        }
-
-        @Override
-        public int getRemoteEdgeDestinations(List<AbstractPort> result) {
-            result.clear();
-            for (int i = 0; i < connections.size(); i++) {
-                ProxyPort pp = remotePortRegister.get(connections.get(i).handle);
-                if (pp != null) {
-                    result.add(pp.getPort());
-                }
-            }
-            int numberOfReverseConnections = 0;
-            if (getExtraEdgeProvider() != null) {
-                numberOfReverseConnections = getExtraEdgeProvider().getRemoteEdgeDestinations(result);
-            }
-            for (FrameworkElementInfo.NetworkConnection connection : networkConnections) {
-                AbstractPort port = connection.getDestinationPort(currentModelNode);
-                if (port != null) {
-                    if (connection.isDestinationSource()) {
-                        numberOfReverseConnections++;
-                        result.add(port);
-                    } else {
-                        result.add(result.size() - numberOfReverseConnections, port);
-                    }
-                }
-            }
-            return result.size() - numberOfReverseConnections;
-        }
-
-        @Override
-        protected void postChildInit() {
-            this.connection = (getPort().getFlag(FrameworkElementFlags.EXPRESS_PORT) ||
-                               (getPort().getDataType().getJavaClass() != null && CCType.class.isAssignableFrom(getPort().getDataType().getJavaClass()))) ? expressConnection : bulkConnection;
-            super.postChildInit();
-        }
-    }
-
-    /**
-     * (Belongs to ProxyPort)
-     *
-     * Create Port Creation info from PortInfo class.
-     * Except from Shared flag port will be identical to original port.
-     *
-     * @param portInfo Port Information
-     * @return Port Creation info
-     */
-    private static PortCreationInfo createPCI(FrameworkElementInfo portInfo) {
-        PortCreationInfo pci = new PortCreationInfo(portInfo.getFlags());
-        pci.flags = portInfo.getFlags();
-
-        // set queue size
-        pci.maxQueueSize = portInfo.getStrategy();
-
-        pci.dataType = portInfo.getDataType();
-        pci.lockOrder = LockOrderLevels.REMOTE_PORT;
-
-        return pci;
-    }
-
-    /**
-     * Pull call storage and management
-     */
-    class PullCall extends SerializedTCPCommand {
-
-        PullCall(NetPort netport) {
-            super(TCP.OpCode.PULLCALL, 16);
-            timeSent = System.currentTimeMillis();
-            callId = nextCallId.incrementAndGet();
-            encoding = netport.getNetworkEncoding();
-            getWriteStream().writeInt(netport.getRemoteHandle());
-            getWriteStream().writeLong(callId);
-            getWriteStream().writeEnum(encoding);
-        }
-
-        /** Time when call was created/sent */
-        final long timeSent;
-
-        /** Call id of pull call */
-        final long callId;
-
-        /** Port call originates from - in case of a standard port */
-        PortBase origin;
-
-        /** Buffer with result */
-        PortDataManager resultBuffer;
-
-        /** Result buffer for CC port */
-        CCPortDataManagerTL ccResultBuffer;
-
-        /** Was CC Pull successful? */
-        boolean ccPullSuccess;
-
-        /** Data encoding to use */
-        Serialization.DataEncoding encoding;
+    public ModelHandler getModelHandler() {
+        return peerImplementation.connectionElement.getModelHandler();
     }
 }

internal/TCP.java

@@ -29 +29 @@
 import org.finroc.core.RuntimeEnvironment;
 import org.finroc.core.RuntimeSettings;
-import org.finroc.core.datatype.FrameworkElementInfo;
+import org.finroc.core.net.generic_protocol.Definitions;
 import org.finroc.core.parameter.ConstructorParameters;
 import org.finroc.core.parameter.StaticParameterList;
@@ -45 +45 @@
  * Plugin for P2P TCP connections
  */
-public class TCP implements Plugin {
+public class TCP extends Definitions implements Plugin {
 
     /** Singleton instance of TCP plugin */
@@ -58 +58 @@
 
     /**
-     * Protocol OpCodes
+     * Flags for connection properties
      */
-    public enum OpCode {
-
-        // Opcodes for management connection
-        SUBSCRIBE,         // Subscribe to data port
-        UNSUBSCRIBE,       // Unsubscribe from data port
-        PULLCALL,          // Pull call
-        PULLCALL_RETURN,   // Returning pull call
-        RPC_CALL,          // RPC call
-        TYPE_UPDATE,       // Update on remote type info (typically desired update time)
-        STRUCTURE_CREATE,  // Update on remote framework elements: Element created
-        STRUCTURE_CHANGE,  // Update on remote framework elements: Port changed
-        STRUCTURE_DELETE,  // Update on remote framework elements: Element deleted
-        PEER_INFO,         // Information about other peers
-
-        // Change event opcodes (from subscription - or for plain setting of port)
-        PORT_VALUE_CHANGE,                         // normal variant
-        SMALL_PORT_VALUE_CHANGE,                   // variant with max. 256 byte message length (3 bytes smaller than normal variant)
-        SMALL_PORT_VALUE_CHANGE_WITHOUT_TIMESTAMP, // variant with max. 256 byte message length and no timestamp (11 bytes smaller than normal variant)
-
-        // Used for messages without opcode
-        OTHER
-    }
-
-    public enum MessageSize {
-        FIXED,                   // fixed size message
-        VARIABLE_UP_TO_255_BYTE, // variable message size up to 255 bytes
-        VARIABLE_UP_TO_4GB       // variable message size up to 4GB
-    };
+    public static final int
+    PRIMARY_CONNECTION = 0x1,  //<! This connection is used to transfer management data (e.g. available ports, peer info, subscriptions, rpc calls)
+    EXPRESS_DATA    = 0x2,  //<! This connection transfers "express data" (port values of express ports) - candidate for UDP in the future
+    BULK_DATA       = 0x4,  //<! This connection transfers "bulk data" (port values of bulk ports) - candidate for UDP in the future
+    JAVA_PEER       = 0x8,  //<! Sent by Java Peers on connection initialization
+    NO_DEBUG        = 0x10; //<! No debug information in protocol
 
     /** Mode of peer */
@@ -95 +73 @@
         FULL          // Peer is client and server
     }
-
-    /**
-     * Flags for connection properties
-     */
-    public static final int
-    MANAGEMENT_DATA = 0x1, //<! This connection is used to transfer management data (e.g. available ports, peer info, subscriptions, rpc calls)
-    EXPRESS_DATA    = 0x2, //<! This connection transfers "express data" (port values of express ports) - candidate for UDP in the future
-    BULK_DATA       = 0x4; //<! This connection transfers "bulk data" (port values of bulk ports) - candidate for UDP in the future
-
-    /** Message size encodings of different kinds of op codes */
-    public static final MessageSize[] MESSAGE_SIZES = new MessageSize[] {
-        MessageSize.FIXED,                   // SUBSCRIBE
-        MessageSize.FIXED,                   // UNSUBSCRIBE
-        MessageSize.FIXED,                   // PULLCALL
-        MessageSize.VARIABLE_UP_TO_4GB,      // PULLCALL_RETURN
-        MessageSize.VARIABLE_UP_TO_4GB,      // RPC_CALL
-        MessageSize.VARIABLE_UP_TO_4GB,      // UPDATE_TIME
-        MessageSize.VARIABLE_UP_TO_4GB,      // STRUCTURE_CREATE
-        MessageSize.VARIABLE_UP_TO_4GB,      // STRUCTURE_CHANGE
-        MessageSize.FIXED,                   // STRUCTURE_DELETE
-        MessageSize.VARIABLE_UP_TO_4GB,      // PEER_INFO
-        MessageSize.VARIABLE_UP_TO_4GB,      // PORT_VALUE_CHANGE
-        MessageSize.VARIABLE_UP_TO_255_BYTE, // SMALL_PORT_VALUE_CHANGE
-        MessageSize.VARIABLE_UP_TO_255_BYTE, // SMALL_PORT_VALUE_CHANGE_WITHOUT_TIMESTAMP
-        MessageSize.VARIABLE_UP_TO_4GB,      // OTHER
-    };
 
     /** Return Status */
@@ -131 +83 @@
     public static final String GREET_MESSAGE = "Greetings! I am a Finroc TCP peer.";
 
-    /** Finroc TCP protocol version */
-    public static final short PROTOCOL_VERSION = 1;
-
-    /** Inserted at the end of messages to debug TCP stream */
-    public static final byte DEBUG_TCP_NUMBER = (byte)0xCD;
-
     /** Standard TCP connection creator */
-    public static final CreateAction creator1 = new CreateAction(FrameworkElementInfo.StructureExchange.COMPLETE_STRUCTURE, "TCP", 0);
+    public static final CreateAction creator1 = new CreateAction(Definitions.StructureExchange.COMPLETE_STRUCTURE, "TCP", 0);
 
     /** Alternative TCP connection creator */
-    public static final CreateAction creator2 = new CreateAction(FrameworkElementInfo.StructureExchange.SHARED_PORTS, TCP_PORTS_ONLY_NAME, 0);
+    public static final CreateAction creator2 = new CreateAction(Definitions.StructureExchange.SHARED_PORTS, TCP_PORTS_ONLY_NAME, 0);
 
     /** Complete TCP connection creator */
-    public static final CreateAction creator3 = new CreateAction(FrameworkElementInfo.StructureExchange.FINSTRUCT, "TCP finstruct", CreateExternalConnectionAction.REMOTE_EDGE_INFO);
+    public static final CreateAction creator3 = new CreateAction(Definitions.StructureExchange.FINSTRUCT, "TCP finstruct", CreateExternalConnectionAction.REMOTE_EDGE_INFO);
 
 
@@ -198 +144 @@
 
         /** Desired structure exchange */
-        private final FrameworkElementInfo.StructureExchange structureExchange;
+        private final Definitions.StructureExchange structureExchange;
 
         /** Name of connection type */
@@ -209 +155 @@
         private final String group;
 
-        public CreateAction(FrameworkElementInfo.StructureExchange structureExchange, String name, int flags) {
+        public CreateAction(Definitions.StructureExchange structureExchange, String name, int flags) {
             this.structureExchange = structureExchange;
             this.name = name;

internal/TCPConnection.java

@@ -35 +35 @@
 import org.rrlib.finroc_core_utils.jc.AtomicDoubleInt;
 import org.rrlib.finroc_core_utils.jc.HasDestructor;
-import org.rrlib.finroc_core_utils.jc.MutexLockOrder;
-import org.rrlib.finroc_core_utils.jc.Time;
-import org.rrlib.finroc_core_utils.jc.container.SafeConcurrentlyIterableList;
 import org.rrlib.finroc_core_utils.jc.container.WonderQueue;
 import org.rrlib.logging.Log;
@@ -45 +42 @@
 import org.rrlib.serialization.InputStreamSource;
 import org.rrlib.serialization.MemoryBuffer;
-import org.rrlib.serialization.Serialization;
+import org.rrlib.serialization.SerializationInfo;
 import org.rrlib.serialization.rtti.DataTypeBase;
 
-import org.finroc.core.LockOrderLevels;
-import org.finroc.core.RuntimeEnvironment;
-import org.finroc.core.RuntimeSettings;
 import org.finroc.core.datatype.CoreString;
-import org.finroc.core.datatype.FrameworkElementInfo;
-import org.finroc.core.parameter.ParameterNumeric;
+import org.finroc.core.net.generic_protocol.Connection;
+import org.finroc.core.net.generic_protocol.Definitions;
+import org.finroc.core.net.generic_protocol.RemoteProxyPort;
+import org.finroc.core.net.generic_protocol.SerializedMessage;
 import org.finroc.core.port.AbstractPort;
 import org.finroc.core.port.ThreadLocalCache;
-import org.finroc.core.port.net.UpdateTimeChangeListener;
 import org.finroc.core.port.rpc.FutureStatus;
 import org.finroc.core.port.rpc.internal.AbstractCall;
-import org.finroc.core.port.rpc.internal.ResponseSender;
+import org.finroc.core.remote.BufferedModelChanges;
+import org.finroc.core.remote.ModelHandler;
 import org.finroc.core.remote.ModelNode;
-import org.finroc.core.remote.RemoteTypes;
+import org.finroc.core.remote.RemoteType;
 import org.finroc.core.thread.CoreLoopThreadBase;
 import org.finroc.plugins.tcp.TCPSettings;
-import org.finroc.plugins.tcp.internal.TCP.OpCode;
 
 /**
@@ -73 +68 @@
  * (writer and listener members need to be initialized by subclass)
  */
-class TCPConnection implements UpdateTimeChangeListener, ResponseSender {
+public class TCPConnection extends Connection {
 
     /** Network Socket used for accessing remote Server */
     private final Socket socket;
 
-    /** default connection times of connection partner */
-    final RemoteTypes remoteTypes = new RemoteTypes();
-
     /** Socket's output stream */
     private final OutputStream socketOutputStream;
 
-    /** Buffer for writing data to stream */
-    private final MemoryBuffer writeBuffer = new MemoryBuffer(MemoryBuffer.DEFAULT_SIZE, MemoryBuffer.DEFAULT_RESIZE_FACTOR, false);
-
-    /** Output Stream for sending data to remote Server */
-    private final BinaryOutputStream writeBufferStream = new BinaryOutputStream(writeBuffer, remoteTypes);
-
-    /** Input Stream for receiving data ro remote Server */
-    private final BinaryInputStream readBufferStream;
-
-    /** Reference to remote part this connection belongs to */
-    private final RemotePart remotePart;
-
     /** Writer Thread */
     private Writer writer;
@@ -102 +82 @@
     private Reader reader;
 
-    /** List with calls that wait for a response */
-    final ArrayList<CallAndTimeout> callsAwaitingResponse = new ArrayList<CallAndTimeout>();
-
-    /** References to Connection parameters */
-    private ParameterNumeric<Integer> minUpdateInterval;
-    private ParameterNumeric<Integer> maxNotAcknowledgedPackets;
-
-    /** Index of last acknowledged sent packet */
-    private volatile int lastAcknowledgedPacket = 0;
-
-    /** Index of last acknowledgement request that was received */
-    private volatile int lastAckRequestIndex = 0;
-
-    /** Timestamp of when packet n was sent (Index is n % MAX_NOT_ACKNOWLEDGED_PACKETS => efficient and safe implementation (ring queue)) */
-    private final long[] sentPacketTime = new long[TCPSettings.MAX_NOT_ACKNOWLEDGED_PACKETS + 1];
-
-    /** Ping time for last packages (Index is n % AVG_PING_PACKETS => efficient and safe implementation (ring queue)) */
-    private final int[] pingTimes = new int[TCPSettings.AVG_PING_PACKETS + 1];
-
-    /** Ping time statistics */
-    private volatile int avgPingTime, maxPingTime;
-
-    /** Signal for disconnecting */
-    private volatile boolean disconnectSignal = false;
-
-    /** Connection type - BULK or EXPRESS */
-    protected final boolean bulk;
-
-    /** Ports that are monitored for changes by this connection and should be checked for modifications */
-    protected SafeConcurrentlyIterableList<TCPPort> monitoredPorts = new SafeConcurrentlyIterableList<TCPPort>(50, 4);
-
     /** TCPPeer that this connection belongs to (null if it does not belong to a peer) */
     protected final TCPPeer peer;
@@ -139 +88 @@
     protected int lastPeerInfoSentRevision = -1;
 
-    /** Rx related: last time RX was retrieved */
-    protected long lastRxTimestamp = 0;
-
-    /** Rx related: last time RX was retrieved: how much have we received in total? */
-    protected long lastRxPosition = 0;
-
-    /** Needs to be locked after framework elements, but before runtime registry */
-    public final MutexLockOrder objMutex = new MutexLockOrder(LockOrderLevels.REMOTE + 1);
-
-    /** Log description for connection */
-    protected String description;
-
-    /** Temporary buffer with port information */
-    private final FrameworkElementInfo tempFrameworkElementInfo = new FrameworkElementInfo();
-
+    /** Whether connection has been disconnected - or is disconnected */
+    private volatile boolean disconnected = false;
 
     /**
@@ -169 +105 @@
         this.socket = socket;
         this.socketOutputStream = socket.getOutputStream();
-        this.readBufferStream = new BinaryInputStream(new InputStreamSource(socket.getInputStream()), remoteTypes);
+        this.readBufferStream = new BinaryInputStream(new InputStreamSource(socket.getInputStream()), INITIAL_SERIALIZATION_INFO);
 
         // Initialize connection
         // Write...
         writeBufferStream.writeString(TCP.GREET_MESSAGE);
-        writeBufferStream.writeShort(TCP.PROTOCOL_VERSION);
+        writeBufferStream.writeShort(Definitions.PROTOCOL_VERSION_MAJOR);
         writeBufferStream.writeSkipOffsetPlaceholder();
         // ConnectionInitMessage
@@ -181 +117 @@
         writeBufferStream.writeString(peer.thisPeer.name);
         writeBufferStream.writeEnum(peer.structureExchange);
-        writeBufferStream.writeInt((connectionFlags & TCP.BULK_DATA) != 0 ? TCP.BULK_DATA : (TCP.EXPRESS_DATA | TCP.MANAGEMENT_DATA));
+        writeBufferStream.writeInt(((connectionFlags & TCP.PRIMARY_CONNECTION) != 0 ? (TCP.BULK_DATA | TCP.PRIMARY_CONNECTION) : TCP.EXPRESS_DATA) | TCP.JAVA_PEER | (Definitions.PROTOCOL_VERSION_MINOR << 16));
         TCP.serializeInetAddress(writeBufferStream, socket.getInetAddress());
-        if (TCPSettings.DEBUG_TCP) {
-            writeBufferStream.writeByte(TCP.DEBUG_TCP_NUMBER);
-        }
+        writeBufferStream.writeByte(Definitions.DEBUG_TCP_NUMBER);
         writeBufferStream.skipTargetHere();
         writeBufferToNetwork();
@@ -197 +131 @@
             throw new ConnectException("Partner does not speak Finroc protocol");
         }
-        if (readBufferStream.readShort() != TCP.PROTOCOL_VERSION) {
+        if (readBufferStream.readShort() != Definitions.PROTOCOL_VERSION_MAJOR) {
             Log.log(LogLevel.WARNING, this, "Connection partner has wrong protocol version");
             throw new ConnectException("Partner has wrong protocol version");
@@ -206 +140 @@
         TCP.PeerType peerType = readBufferStream.readEnum(TCP.PeerType.class);
         String peerName = readBufferStream.readString();
-        FrameworkElementInfo.StructureExchange structureExchange = readBufferStream.readEnum(FrameworkElementInfo.StructureExchange.class); // TODO: send structure
-        connectionFlags |= readBufferStream.readInt();
-        this.bulk = (connectionFlags & TCP.BULK_DATA) != 0;
+        Definitions.StructureExchange structureExchange = readBufferStream.readEnum(Definitions.StructureExchange.class); // TODO: send structure
+        int flags = readBufferStream.readInt();
+        int partnerFlags = flags & 0xFF;
+        int partnerSerializationRevision = flags >> 16;
+        boolean legacyPartnerRuntime = partnerSerializationRevision == 0;
+        int partnerHandleStampWidth = legacyPartnerRuntime ? 12 : ((flags >> 8) & 0xFF);
+        connectionFlags |= partnerFlags;
+        this.primary = (connectionFlags & TCP.PRIMARY_CONNECTION) != 0;
         InetAddress myAddress = TCP.deserializeInetAddress(readBufferStream);
-        checkCommandEnd(readBufferStream);
+        boolean debugTcp = legacyPartnerRuntime || (!((connectionFlags & TCP.NO_DEBUG) != 0 || (flags & TCP.NO_DEBUG) != 0));
+        checkCommandEnd(readBufferStream);  // requires serialization info to bet set up
 
         // Adjust peer management information
         synchronized (peer.connectTo) {
             peer.addOwnAddress(myAddress);
-            remotePart = peer.getRemotePart(uuid, peerType, peerName, socket.getInetAddress(), neverForget);
-            remotePart.peerInfo.connecting = true;
-            if (!remotePart.addConnection(this)) {
+            remoteRuntime = peer.getRemotePart(uuid, peerType, peerName, socket.getInetAddress(), neverForget, partnerHandleStampWidth);
+            ((RemotePart)remoteRuntime).peerInfo.connecting = true;
+            if (!remoteRuntime.addConnection(this)) {
                 throw new ConnectException("Connection already established. Closing.");
             }
-            if (peer.thisPeer.peerType != TCP.PeerType.CLIENT_ONLY) {
-                remotePart.setDesiredStructureInfo(structureExchange);
-            }
+        }
+
+        // Setup serialization info
+        int otherSerializationFlags = structureExchange.ordinal() | (debugTcp ? Definitions.INFO_FLAG_DEBUG_PROTOCOL : 0);
+        int otherDeserializationFlags = peer.structureExchange.ordinal() | Definitions.INFO_FLAG_JAVA_CLIENT | (debugTcp ? Definitions.INFO_FLAG_DEBUG_PROTOCOL : 0);
+        boolean javaPartner = (partnerFlags & TCP.JAVA_PEER) != 0;
+        boolean primaryConnection = (partnerFlags & TCP.PRIMARY_CONNECTION) != 0;
+        boolean expressOnlyConnection = (partnerFlags & TCP.EXPRESS_DATA) != 0 && (!primaryConnection) || (connectionFlags & TCP.EXPRESS_DATA) != 0;
+        if (legacyPartnerRuntime) {
+            SerializationInfo legacySerialization = new SerializationInfo(0, INITIAL_SERIALIZATION_INFO.getRegisterEntryEncodings(), otherSerializationFlags);
+            writeBufferStream.setSerializationTarget(legacySerialization);
+            SerializationInfo legacyDeserialization = new SerializationInfo(0, INITIAL_SERIALIZATION_INFO.getRegisterEntryEncodings(), otherDeserializationFlags);
+            readBufferStream.setSerializationSource(legacyDeserialization);
+        } else if (expressOnlyConnection) {
+            writeBufferStream.setSharedSerializationInfo(remoteRuntime.getPrimaryConnection().getWriteBufferStream());
+            readBufferStream.setSharedSerializationInfo(remoteRuntime.getPrimaryConnection().getReadBufferStream());
+        } else {
+            int revision = Math.min(Definitions.PROTOCOL_VERSION_MINOR, partnerSerializationRevision);
+            int writeEncoding = INITIAL_SERIALIZATION_INFO.getRegisterEntryEncodings();
+            int javaEncoding = SerializationInfo.setRegisterEntryEncoding(SerializationInfo.setDefaultRegisterEntryEncoding(SerializationInfo.RegisterEntryEncoding.PUBLISH_REGISTER_ON_CHANGE, Definitions.RegisterUIDs.values().length), Definitions.RegisterUIDs.CREATE_ACTION.ordinal(), SerializationInfo.RegisterEntryEncoding.PUBLISH_REGISTER_ON_DEMAND);
+            if (javaPartner) {
+                writeEncoding = javaEncoding;
+                otherSerializationFlags |= Definitions.INFO_FLAG_JAVA_CLIENT;
+            }
+            writeBufferStream.setSerializationTarget(new SerializationInfo(revision, writeEncoding, otherSerializationFlags));
+            readBufferStream.setSerializationSource(new SerializationInfo(revision, javaEncoding, otherDeserializationFlags));
         }
 
         // set params
-        minUpdateInterval = bulk ? TCPSettings.getInstance().minUpdateIntervalBulk : TCPSettings.getInstance().minUpdateIntervalExpress;
-        maxNotAcknowledgedPackets = bulk ? TCPSettings.getInstance().maxNotAcknowledgedPacketsBulk : TCPSettings.getInstance().maxNotAcknowledgedPacketsExpress;
+        minUpdateInterval = primary ? TCPSettings.getInstance().minUpdateIntervalBulk : TCPSettings.getInstance().minUpdateIntervalExpress;
+        maxNotAcknowledgedPackets = primary ? TCPSettings.getInstance().maxNotAcknowledgedPacketsBulk : TCPSettings.getInstance().maxNotAcknowledgedPacketsExpress;
 
         // Init connection
         readBufferStream.setTimeout(-1);
-        String typeString = bulk ? "Bulk" : "Express";
+        String typeString = primary ? "Primary" : "Express";
 
         // Initialize them here, so that subscribe calls from structure creation do not get lost
@@ -238 +201 @@
 
         // Get framework elements from connection partner
-        if ((connectionFlags & TCP.MANAGEMENT_DATA) != 0) {
-            peer.connectionElement.getModelHandler().changeNodeName(modelNode, "Connecting to " + remotePart.peerInfo.toString() + "...");
+        if ((connectionFlags & TCP.PRIMARY_CONNECTION) != 0) {
+            BufferedModelChanges change = new BufferedModelChanges();
+            change.changeNodeName(modelNode, "Connecting to " + ((RemotePart)remoteRuntime).peerInfo.toString() + "...");
+            getModelHandler().applyModelChanges(change);
+
+
             //ModelNode statusNode = new ModelNode("Obtaining structure information...");
             //peer.connectionElement.getModelHandler().addNode(modelNode, statusNode);
 
             // retrieveRemotePorts(cis, cos, updateTimes, newServer);
-            boolean newServer = true; /*(serverCreationTime < 0) || (serverCreationTime != timeBase);*/
+            //boolean newServer = true; /*(serverCreationTime < 0) || (serverCreationTime != timeBase);*/
             //log(LogLevel.LL_DEBUG, this, (newServer ? "Connecting" : "Reconnecting") + " to server " + uuid.toString() + "...");
 
-            // Delete any elements from previous connections
-            remotePart.deleteAllChildren();
-            remotePart.createNewModel();
+            // Delete any elements from previous connections (no longer necessary)
+            //remoteRuntime.deleteAllChildren();
+            remoteRuntime.createNewModel();
 
             /*portIterator.reset();
@@ -264 +231 @@
                 // Process structure packets
                 int structurePacketSize = readBufferStream.readInt();
-                boolean readType = peer.structureExchange == FrameworkElementInfo.StructureExchange.SHARED_PORTS;
+                boolean readType = peer.structureExchange == Definitions.StructureExchange.SHARED_PORTS;
                 MemoryBuffer structurePacketBuffer = new MemoryBuffer(structurePacketSize);
-                BinaryInputStream structurePacketReadStream = new BinaryInputStream(structurePacketBuffer, remoteTypes);
+                BinaryInputStream structurePacketReadStream = new BinaryInputStream(structurePacketBuffer, readBufferStream);
                 while (structurePacketSize != 0) {
                     structurePacketBuffer.setSize(structurePacketSize);
@@ -272 +239 @@
                     structurePacketReadStream.reset();
                     if (readType) {
-                        DataTypeBase type = structurePacketReadStream.readType();
+                        DataTypeBase type = RemoteType.deserialize(structurePacketReadStream).getDefaultLocalDataType();
                         if (type == null || type != CoreString.TYPE) {
                             Log.log(LogLevel.WARNING, this, "Type encoding does not seem to work");
@@ -280 +247 @@
                     }
                     while (structurePacketReadStream.moreDataAvailable()) {
-                        tempFrameworkElementInfo.deserialize(structurePacketReadStream, peer.structureExchange);
-                        remotePart.addRemoteStructure(tempFrameworkElementInfo, true);
+                        tempFrameworkElementInfo.deserialize(structurePacketReadStream, true);
+                        remoteRuntime.addRemoteStructure(tempFrameworkElementInfo, true, null);
                     }
                     structurePacketSize = readBufferStream.readInt();
                 }
-                // remotePart.initAndCheckForAdminPort(modelNode); do this later: after bulk connection has been initialized
+                // remoteRuntime.initAndCheckForAdminPort(modelNode); do this later: after bulk connection has been initialized
             } catch (Exception e) {
                 if (e.getCause() instanceof EOFException) {
@@ -293 +260 @@
                 }
                 //peer.connectionElement.getModelHandler().removeNode(statusNode);
-                remotePart.removeConnection(this);
+                remoteRuntime.removeConnection(this);
                 throw e;
             }
@@ -309 +276 @@
     }
 
-    /**
-     * @return Is TCP connection disconnecting?
-     */
-    public boolean disconnecting() {
-        return disconnectSignal;
-    }
-
-    /**
-     * @return Type of connection ("Bulk" oder "Express")
-     */
-    public String getConnectionTypeString() {
-        return bulk ? "Bulk" : "Express";
-    }
-
-    /**
-     * Close connection
-     */
+    @Override
     public synchronized void disconnect() {
         disconnectSignal = true;
-        RuntimeSettings.getInstance().removeUpdateTimeChangeListener(this);
         synchronized (peer.connectTo) {
-            remotePart.removeConnection(this);
+            remoteRuntime.removeConnection(this);
         }
         notifyWriter(); // stops writer
@@ -360 +310 @@
 
     /**
-     * Check that command is terminated correctly when TCPSettings.DEBUG_TCP is activated
-     */
-    static void checkCommandEnd(BinaryInputStream stream) {
-        if (TCPSettings.DEBUG_TCP) {
-            int i = stream.readByte();
-            if (i != TCP.DEBUG_TCP_NUMBER) {
-                throw new RuntimeException("TCP Stream seems corrupt");
-            }
-        }
-    }
-
-    /**
      * Listens at socket for incoming data
      */
@@ -386 +324 @@
             initThreadLocalCache();
             BinaryInputStream stream = TCPConnection.this.readBufferStream;
-            MemoryBuffer structureBuffer = new MemoryBuffer(); // structure changes are copied to this buffer and processed by model (thread)
-            BinaryOutputStream structureBufferWriter = new BinaryOutputStream(structureBuffer);
-            byte[] tempBuffer = new byte[2048];
+            BufferedModelChanges modelChanges = new BufferedModelChanges();
 
             try {
@@ -410 +346 @@
                         while (lastAcknowledgedPacket != acknowledgement) {
                             lastAcknowledgedPacket = (lastAcknowledgedPacket + 1) & 0x7FFF;
-                            pingTimes[lastAcknowledgedPacket & TCPSettings.AVG_PING_PACKETS] =
-                                (int)(curTime - sentPacketTime[lastAcknowledgedPacket & TCPSettings.MAX_NOT_ACKNOWLEDGED_PACKETS]);
+                            pingTimes[lastAcknowledgedPacket & Definitions.AVG_PING_PACKETS] =
+                                (int)(curTime - sentPacketTime[lastAcknowledgedPacket & Definitions.MAX_NOT_ACKNOWLEDGED_PACKETS]);
                         }
                         updatePingStatistics();
@@ -417 +353 @@
 
                     while (stream.getAbsoluteReadPosition() < nextPacketStart) {
-                        TCP.OpCode opCode = stream.readEnum(TCP.OpCode.class);
-                        if (opCode.ordinal() >= TCP.OpCode.OTHER.ordinal()) {
+                        Definitions.OpCode opCode = stream.readEnum(Definitions.OpCode.class);
+                        if (opCode.ordinal() >= Definitions.OpCode.OTHER.ordinal()) {
                             Log.log(LogLevel.WARNING, this, "Received corrupted TCP message batch. Invalid opcode. Skipping.");
                             stream.skip((int)(nextPacketStart - stream.getAbsoluteReadPosition()));
                             break;
                         }
-                        TCP.MessageSize messageSizeEncoding = TCP.MESSAGE_SIZES[opCode.ordinal()];
-                        long messageSize = messageSizeEncoding == TCP.MessageSize.FIXED ? -1 :
-                                           (messageSizeEncoding == TCP.MessageSize.VARIABLE_UP_TO_255_BYTE ? (stream.readByte() & 0xFF) : stream.readInt());
+                        Definitions.MessageSize messageSizeEncoding = Definitions.MESSAGE_SIZES[opCode.ordinal()];
+                        long messageSize = messageSizeEncoding == Definitions.MessageSize.FIXED ? -1 :
+                                           (messageSizeEncoding == Definitions.MessageSize.VARIABLE_UP_TO_255_BYTE ? (stream.readByte() & 0xFF) : stream.readInt());
                         //long messageEncodingSize = messageSizeEncoding.ordinal() * messageSizeEncoding.ordinal(); // :-)
                         long commandStartPosition = stream.getAbsoluteReadPosition();
@@ -436 +372 @@
 
                         // Copy to structure buffer?
-                        if (opCode == OpCode.STRUCTURE_CREATE || opCode == OpCode.STRUCTURE_CHANGE || opCode == OpCode.STRUCTURE_DELETE) {
+                        /*if (opCode == Definitions.OpCode.STRUCTURE_CREATED || opCode == Definitions.OpCode.STRUCTURE_CHANGED || opCode == Definitions.OpCode.STRUCTURE_DELETED) {
+                            remoteRuntime.processStructure
+                            // TODO because of register entries this needs to be read now
                             structureBufferWriter.writeEnum(opCode);
-                            long remaining = (opCode == OpCode.STRUCTURE_DELETE) ? (TCPSettings.DEBUG_TCP ? 5 : 4) : messageSize;
+                            long remaining = (opCode == Definitions.OpCode.STRUCTURE_DELETED) ? (TCPSettings.DEBUG_TCP ? 5 : 4) : messageSize;
                             while (remaining > 0) {
                                 long copy = Math.min(tempBuffer.length, remaining);
@@ -445 +383 @@
                                 remaining -= copy;
                             }
-                        } else {
-                            try {
-                                remotePart.processMessage(opCode, stream, remoteTypes, TCPConnection.this);
-                                checkCommandEnd(stream);
-                            } catch (Exception e) {
-                                Log.log(LogLevel.WARNING, this, "Failed to deserialize message of type " + opCode.toString() + ". Skipping. Reason: ", e);
-                                long skip = nextCommandStartPosition - stream.getAbsoluteReadPosition();
+                        } else {*/
+                        try {
+                            remoteRuntime.processMessage(opCode, stream, TCPConnection.this, modelChanges);
+                            checkCommandEnd(stream);
+                        } catch (Exception e) {
+                            Log.log(LogLevel.WARNING, this, "Failed to deserialize message of type " + opCode.toString() + ". Skipping. Reason: ", e);
+                            long skip = nextCommandStartPosition - stream.getAbsoluteReadPosition();
+                            if (skip >= 0) {
+                                stream.skip((int)skip);
+                            } else {
+                                skip = nextPacketStart - stream.getAbsoluteReadPosition();
                                 if (skip >= 0) {
-                                    stream.skip((int)skip);
+                                    Log.log(LogLevel.WARNING, this, "Too much stream content was read. This is not handled yet. Moving to next message batch."); // TODO
                                 } else {
-                                    skip = nextPacketStart - stream.getAbsoluteReadPosition();
-                                    if (skip >= 0) {
-                                        Log.log(LogLevel.WARNING, this, "Too much stream content was read. This is not handled yet. Moving to next message batch."); // TODO
-                                    } else {
-                                        Log.log(LogLevel.WARNING, this, "Too much stream content was read - exceeding message batch. This is not handled yet. Disconnecting."); // TODO
-                                    }
+                                    Log.log(LogLevel.WARNING, this, "Too much stream content was read - exceeding message batch. This is not handled yet. Disconnecting."); // TODO
                                 }
                             }
+                            //}
                         }
                     }
 
+                    if (!modelChanges.empty()) {
+                        getModelHandler().applyModelChanges(modelChanges);
+                        modelChanges = new BufferedModelChanges();
+                    }
+
+                    /*
                     structureBufferWriter.flush();
                     if (structureBuffer.getSize() > 0) {
@@ -475 +419 @@
                             @Override
                             public void run() {
-                                remotePart.processStructurePacket(structureBufferToProcess, remoteTypes);
+                                remoteRuntime.processStructurePacket(structureBufferToProcess, remoteTypes);
                             }
                         });
-                    }
+                    }*/
                 }
             } catch (Exception e) {
+                disconnected = true;
                 if (e instanceof RuntimeException && e.getCause() != null) {
                     e = (Exception)e.getCause();
@@ -489 +434 @@
             }
             try {
-                remotePart.disconnect();
+                remoteRuntime.disconnect();
             } catch (Exception e) {
                 Log.log(LogLevel.WARNING, this, e);
@@ -497 +442 @@
                 stream.close();
             } catch (Exception e) {}
+            disconnected = true;
         }
 
@@ -525 +471 @@
 //    }
 
-    /**
-     * Notify (possibly wake-up) writer thread. Should be called whenever new tasks for the writer arrive.
-     */
+    @Override
     public void notifyWriter() {
         Writer lockedWriter = writer;
         if (lockedWriter != null) {
             lockedWriter.notifyWriter();
-        }
-    }
-
-    /** Class to store a pair - call and timeout - in a list */
-    static class CallAndTimeout {
-
-        long timeoutTime;
-        AbstractCall call;
-
-        public CallAndTimeout(long timeoutTime, AbstractCall call) {
-            this.timeoutTime = timeoutTime;
-            this.call = call;
         }
     }
@@ -574 +506 @@
 
         /** Queue with TCP commands waiting to be sent */
-        private final WonderQueue<SerializedTCPCommand> tcpCallsToSend = new WonderQueue<SerializedTCPCommand>();
+        private final WonderQueue<SerializedMessage> tcpCallsToSend = new WonderQueue<SerializedMessage>();
 
         /** List with calls that were not ready for sending yet */
@@ -610 +542 @@
         public boolean canSend() {
             int maxNotAck = maxNotAcknowledgedPackets.getValue();
-            return curPacketIndex < lastAcknowledgedPacket + maxNotAck || (maxNotAck <= 0 && curPacketIndex < lastAcknowledgedPacket + TCPSettings.MAX_NOT_ACKNOWLEDGED_PACKETS);
+            return curPacketIndex < lastAcknowledgedPacket + maxNotAck || (maxNotAck <= 0 && curPacketIndex < lastAcknowledgedPacket + Definitions.MAX_NOT_ACKNOWLEDGED_PACKETS);
         }
 
@@ -626 +558 @@
                     if (disconnectSignal) {
                         try {
-                            remotePart.disconnect();
+                            remoteRuntime.disconnect();
                         } catch (Exception e) {
                             Log.log(LogLevel.WARNING, this, e);
@@ -679 +611 @@
 
                     // send port data
-                    ArrayWrapper<TCPPort> it = monitoredPorts.getIterable();
+                    ArrayWrapper<RemoteProxyPort> it = monitoredPorts.getIterable();
                     byte changedFlag = 0;
                     for (int i = 0, n = it.size(); i < n; i++) {
-                        TCPPort pp = it.get(i);
+                        RemoteProxyPort pp = it.get(i);
                         if (pp != null && pp.getPort().isReady()) {
                             if (pp.getLastUpdate() + pp.getUpdateIntervalForNet() > startTime) {
@@ -737 +669 @@
                         writeBuffer.getBuffer().putShort(4, curPacketIndex & 0x7FFF); // TODO
                         lastAcknowledgementRequestTime = System.currentTimeMillis();
-                        sentPacketTime[curPacketIndex & TCPSettings.MAX_NOT_ACKNOWLEDGED_PACKETS] = lastAcknowledgementRequestTime;
+                        sentPacketTime[curPacketIndex & Definitions.MAX_NOT_ACKNOWLEDGED_PACKETS] = lastAcknowledgementRequestTime;
                     }
 
@@ -752 +684 @@
                 }
             } catch (Exception e) {
-
+                disconnected = true;
                 if (e instanceof RuntimeException && e.getCause() != null) {
                     e = (Exception)e.getCause();
@@ -761 +693 @@
 
                 try {
-                    remotePart.disconnect();
+                    remoteRuntime.disconnect();
                 } catch (Exception e2) {
                     Log.log(LogLevel.WARNING, this, e2);
@@ -770 +702 @@
                 stream.close();
             } catch (Exception e) {}
+            disconnected = true;
         }
 
@@ -846 +779 @@
             }
 
-            SerializedTCPCommand tcpCall = null;
+            SerializedMessage tcpCall = null;
             while ((tcpCall = tcpCallsToSend.dequeue()) != null) {
                 tcpCall.serialize(stream);
@@ -874 +807 @@
             boolean expectsResponse = call.expectsResponse();
             if (expectsResponse) {
-                call.setCallId(remotePart.nextCallId.incrementAndGet());
-            }
-
-            writeBufferStream.writeEnum(TCP.OpCode.RPC_CALL);
+                call.setCallId(remoteRuntime.getUniqueCallId());
+            }
+
+            writeBufferStream.writeEnum(Definitions.OpCode.RPC_CALL);
             writeBufferStream.writeSkipOffsetPlaceholder();
             writeBufferStream.writeInt(call.getRemotePortHandle());
@@ -883 +816 @@
             call.serialize(writeBufferStream);
             if (TCPSettings.DEBUG_TCP) {
-                writeBufferStream.writeByte(TCP.DEBUG_TCP_NUMBER);
+                writeBufferStream.writeByte(Definitions.DEBUG_TCP_NUMBER);
             }
             writeBufferStream.skipTargetHere();
@@ -943 +876 @@
          * @param call Call object
          */
-        public void sendCall(SerializedTCPCommand call) {
+        public void sendCall(SerializedMessage call) {
             //call.responsibleThread = -1;
             tcpCallsToSend.enqueue(call);
@@ -949 +882 @@
         }
     }
-
-    /**
-     * Checks whether any waiting calls have timed out.
-     * Removes any timed out calls from list.
-     *
-     * @return Are still calls waiting?
-     */
-    public boolean checkWaitingCallsForTimeout(long timeNow) {
-        synchronized (callsAwaitingResponse) {
-            for (int i = 0; i < callsAwaitingResponse.size(); i++) {
-                if (timeNow > callsAwaitingResponse.get(i).timeoutTime) {
-                    callsAwaitingResponse.get(i).call.setException(FutureStatus.TIMEOUT);
-                    callsAwaitingResponse.remove(i);
-                    i--;
-                }
-            }
-            return callsAwaitingResponse.size() > 0;
-        }
-    }
-
-    /**
-     * Should be called regularly by monitoring thread to check whether critical
-     * ping time threshold is exceeded.
-     *
-     * @return Time the calling thread may wait before calling again (it futile to call this method before)
-     */
-    public long checkPingForDisconnect() {
-        Writer lockedWriter = writer;
-        if (lockedWriter == null) {
-            return TCPSettings.getInstance().criticalPingThreshold.getValue();
-        }
-        if (lastAcknowledgedPacket != lockedWriter.curPacketIndex) {
-            long criticalPacketTime = sentPacketTime[(lastAcknowledgedPacket + 1) & TCPSettings.MAX_NOT_ACKNOWLEDGED_PACKETS];
-            long timeLeft = criticalPacketTime + TCPSettings.getInstance().criticalPingThreshold.getValue() - System.currentTimeMillis();
-            if (timeLeft < 0) {
-                handlePingTimeExceed();
-                return TCPSettings.getInstance().criticalPingThreshold.getValue();
-            }
-            return timeLeft;
-        } else {
-            return TCPSettings.getInstance().criticalPingThreshold.getValue();
-        }
-    }
-
-    /**
-     * Updates ping statistic variables
-     */
-    private void updatePingStatistics() {
-        int result = 0;
-        int resultAvg = 0;
-        for (int i = 0; i < pingTimes.length; i++) {
-            result = Math.max(result, pingTimes[i]);
-            resultAvg += pingTimes[i];
-        }
-        maxPingTime = result;
-        avgPingTime = resultAvg / pingTimes.length;
-    }
-
-    /**
-     * @return Maximum ping time among last TCPSettings.AVG_PING_PACKETS packets
-     */
-    public int getMaxPingTime() {
-        return maxPingTime;
-    }
-
-    /**
-     * @return Average ping time among last TCPSettings.AVG_PING_PACKETS packets
-     */
-    public int getAvgPingTime() {
-        return avgPingTime;
-    }
-
-    /**
-     * @return Is critical ping time currently exceeded (possibly temporary disconnect)
-     */
-    public boolean pingTimeExceeed() {
-        Writer lockedWriter = writer;
-        if (lockedWriter == null) {
-            return false;
-        }
-        if (lastAcknowledgedPacket != lockedWriter.curPacketIndex) {
-            long criticalPacketTime = sentPacketTime[(lastAcknowledgedPacket + 1) & TCPSettings.MAX_NOT_ACKNOWLEDGED_PACKETS];
-            long timeLeft = criticalPacketTime + TCPSettings.getInstance().criticalPingThreshold.getValue() - System.currentTimeMillis();
-            return timeLeft < 0;
-        } else {
-            return false;
-        }
-    }
-
-    /**
-     * Called when critical ping time threshold was exceeded
-     */
-    public void handlePingTimeExceed() {} // TODO
 
 //    /**
@@ -1092 +932 @@
 //    }
 
-    /**
-     * Send call to connection partner
-     *
-     * @param call Call object
-     */
+    @Override
     public void sendCall(AbstractCall call) {
         Writer lockedWriter = writer;
@@ -1106 +942 @@
     }
 
-    /**
-     * Send serialized TCP call to connection partner
-     *
-     * @param call Call object
-     */
-    public void sendCall(SerializedTCPCommand call) {
+    @Override
+    public void sendCall(SerializedMessage call) {
         Writer lockedWriter = writer;
         if (lockedWriter != null && (!disconnectSignal)) {
@@ -1118 +950 @@
     }
 
+    /**
+     * @return Is critical ping time currently exceeded (possibly temporary disconnect)
+     */
     @Override
-    public void updateTimeChanged(DataTypeBase dt, short newUpdateTime) {
-        // forward update time change to connection partner
-        SerializedTCPCommand command = new SerializedTCPCommand(TCP.OpCode.TYPE_UPDATE, 8);
-        command.getWriteStream().writeShort(dt == null ? -1 : dt.getUid());
-        command.getWriteStream().writeShort(newUpdateTime);
-        sendCall(command);
+    public boolean pingTimeExceeed() {
+        Writer lockedWriter = writer;
+        if (lockedWriter == null) {
+            return false;
+        }
+        if (lastAcknowledgedPacket != lockedWriter.curPacketIndex) {
+            long criticalPacketTime = sentPacketTime[(lastAcknowledgedPacket + 1) & Definitions.MAX_NOT_ACKNOWLEDGED_PACKETS];
+            long timeLeft = criticalPacketTime + TCPSettings.getInstance().criticalPingThreshold.getValue() - System.currentTimeMillis();
+            return timeLeft < 0;
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * Should be called regularly by monitoring thread to check whether critical
+     * ping time threshold is exceeded.
+     *
+     * @return Time the calling thread may wait before calling again (it futile to call this method before)
+     */
+    public long checkPingForDisconnect() {
+        Writer lockedWriter = writer;
+        if (lockedWriter == null) {
+            return TCPSettings.getInstance().criticalPingThreshold.getValue();
+        }
+        if (lastAcknowledgedPacket != lockedWriter.curPacketIndex) {
+            long criticalPacketTime = sentPacketTime[(lastAcknowledgedPacket + 1) & Definitions.MAX_NOT_ACKNOWLEDGED_PACKETS];
+            long timeLeft = criticalPacketTime + TCPSettings.getInstance().criticalPingThreshold.getValue() - System.currentTimeMillis();
+            if (timeLeft < 0) {
+                handlePingTimeExceed();
+                return TCPSettings.getInstance().criticalPingThreshold.getValue();
+            }
+            return timeLeft;
+        } else {
+            return TCPSettings.getInstance().criticalPingThreshold.getValue();
+        }
+    }
+
+    /**
+     * Writes all contents of write buffer to network and empties and resets it
+     */
+    protected void writeBufferToNetwork() throws IOException {
+        writeBufferStream.flush();
+        socketOutputStream.write(writeBuffer.getBuffer().getBuffer().array(), 0, writeBuffer.getSize());
+        writeBufferStream.reset();
+    }
+
+    @Override
+    public ModelHandler getModelHandler() {
+        return peer.connectionElement.getModelHandler();
+    }
+
+    /**
+     * @return Whether connection has been disconnected - or is disconnected
+     */
+    public boolean isDisconnected() {
+        return disconnected;
     }
 
@@ -1240 +1126 @@
 //    }
 
-    /**
-     * @return Data rate of bytes read from network (in bytes/s)
-     */
-    public int getRx() {
-        long lastTime = lastRxTimestamp;
-        long lastPos = lastRxPosition;
-        lastRxTimestamp = Time.getCoarse();
-        lastRxPosition = readBufferStream.getAbsoluteReadPosition();
-        if (lastTime == 0) {
-            return 0;
-        }
-        if (lastRxTimestamp == lastTime) {
-            return 0;
-        }
-
-        double data = lastRxPosition - lastPos;
-        double interval = (lastRxTimestamp - lastTime) / 1000;
-        return (int)(data / interval);
-    }
-
-    public String toString() {
-        return description;
-    }
-
-    /**
-     * Writes all contents of write buffer to network and empties and resets it
-     */
-    protected void writeBufferToNetwork() throws IOException {
-        writeBufferStream.flush();
-        socketOutputStream.write(writeBuffer.getBuffer().getBuffer().array(), 0, writeBuffer.getSize());
-        writeBufferStream.reset();
-    }
-
-    /**
-     * Subscribe to port changes on remote server
-     *
-     * @param index Port index in remote runtime
-     * @param strategy Strategy to use/request
-     * @param updateInterval Minimum interval in ms between notifications (values <= 0 mean: use server defaults)
-     * @param localIndex Local Port Index
-     * @param dataType DataType got from server
-     * @param enc Data type to use when writing data to the network
-     */
-    public void subscribe(int index, short strategy, boolean reversePush, short updateInterval, int localIndex, Serialization.DataEncoding enc) {
-        SerializedTCPCommand command = new SerializedTCPCommand(TCP.OpCode.SUBSCRIBE, 16);
-        command.getWriteStream().writeInt(index);
-        command.getWriteStream().writeShort(strategy);
-        command.getWriteStream().writeBoolean(reversePush);
-        command.getWriteStream().writeShort(updateInterval);
-        command.getWriteStream().writeInt(localIndex);
-        command.getWriteStream().writeEnum(enc);
-        sendCall(command);
-    }
-
-    /**
-     * Unsubscribe from port changes on remote server
-     *
-     * @param index Port index in remote runtime
-     */
-    public void unsubscribe(int index) {
-        SerializedTCPCommand command = new SerializedTCPCommand(TCP.OpCode.UNSUBSCRIBE, 8);
-        command.getWriteStream().writeInt(index);
-        sendCall(command);
-    }
-
-    /**
-     * @param portIndex Index of port
-     * @return TCPPort for specified index
-     */
-    protected TCPPort lookupPortForCallHandling(int portIndex) {
-        AbstractPort ap = RuntimeEnvironment.getInstance().getPort(portIndex);
-        TCPPort p = null;
-        if (ap != null) {
-            p = (TCPPort)ap.asNetPort();
-            assert(p != null);
-        }
-        return p;
-    }
-
-    @Override
-    public void sendResponse(AbstractCall responseToSend) {
-        sendCall(responseToSend);
-    }
 }

internal/TCPPeer.java

@@ -36 +36 @@
 import org.finroc.core.FrameworkElement.ChildIterator;
 import org.finroc.core.RuntimeSettings;
-import org.finroc.core.datatype.FrameworkElementInfo;
 import org.finroc.core.plugin.ExternalConnection;
+import org.finroc.core.remote.BufferedModelChanges;
+import org.finroc.core.remote.Definitions;
 import org.finroc.core.remote.ModelNode;
 import org.finroc.plugins.tcp.internal.TCP.PeerType;
@@ -70 +71 @@
 
     /** Amount of structure this client is interested in */
-    final FrameworkElementInfo.StructureExchange structureExchange;
+    final Definitions.StructureExchange structureExchange;
 
     /**
@@ -115 +116 @@
      * @param serverListenAddress The address that server is supposed to listen on ("" will enable IPv6)
      */
-    public TCPPeer(ExternalConnection frameworkElement, String peerName, TCP.PeerType peerType, FrameworkElementInfo.StructureExchange structureExchange, String networkConnection,
+    public TCPPeer(ExternalConnection frameworkElement, String peerName, TCP.PeerType peerType, Definitions.StructureExchange structureExchange, String networkConnection,
                    int preferredServerPort, boolean tryNextPortsIfOccupied, boolean autoConnectToAllPeers, String serverListenAddress) {
         this.connectionElement = frameworkElement;
@@ -244 +245 @@
             activelyConnect = true;
             modelRootNode = new ModelNode("TCP");
-            connectionElement.getModelHandler().setModelRoot(modelRootNode);
+            BufferedModelChanges changes = new BufferedModelChanges();
+            changes.setModelRoot(modelRootNode);
+            connectionElement.getModelHandler().applyModelChanges(changes);
 
             if (address.length() > 0) {
@@ -368 +371 @@
             if (fe instanceof RemotePart && fe.isReady()) {
                 RemotePart rs = (RemotePart)fe;
-                if (!rs.peerInfo.connected) {
+                if (!rs.isConnected()) {
                     return 0;
                 }
@@ -388 +391 @@
                 if (fe instanceof RemotePart && fe.isReady()) {
                     RemotePart rs = (RemotePart)fe;
-                    if (!rs.peerInfo.connected) {
+                    if (!rs.isConnected()) {
                         continue;
                     }
@@ -411 +414 @@
      */
     public boolean isAdminConnection() {
-        return structureExchange == FrameworkElementInfo.StructureExchange.FINSTRUCT;
+        return structureExchange == Definitions.StructureExchange.FINSTRUCT;
     }
 
@@ -445 +448 @@
      * @param address IP address of remote part
      * @param neverForget Is this a remote peer to never forget?
+     * @param partnerHandleStampWidth Stamp Width of partner's framework element handles
      * @return Pointer to remote part
      */
-    public RemotePart getRemotePart(UUID uuid, TCP.PeerType peerType, String peerName, InetAddress address, boolean neverForget) throws Exception {
+    public RemotePart getRemotePart(UUID uuid, TCP.PeerType peerType, String peerName, InetAddress address, boolean neverForget, int partnerHandleStampWidth) throws Exception {
         synchronized (connectTo) {
             if (uuid.equals(this.thisPeer.uuid)) {
@@ -457 +461 @@
                 if (uuid.equals(info.uuid)) {
                     if (info.remotePart == null) {
-                        info.remotePart = new RemotePart(info, connectionElement, this);
+                        info.remotePart = new RemotePart(info, connectionElement, this, partnerHandleStampWidth);
                         info.remotePart.init();
                     }
@@ -475 +479 @@
             info.name = peerName;
             info.neverForget = neverForget;
-            info.remotePart = new RemotePart(info, connectionElement, this);
+            info.remotePart = new RemotePart(info, connectionElement, this, partnerHandleStampWidth);
             info.remotePart.init();
             return info.remotePart;
@@ -511 +515 @@
         }
         if (modelRootNode != null) {
-            connectionElement.getModelHandler().removeNode(modelRootNode);
+            BufferedModelChanges changes = new BufferedModelChanges();
+            changes.removeNode(modelRootNode);
+            connectionElement.getModelHandler().applyModelChanges(changes);
         }
         modelRootNode = null;
@@ -522 +528 @@
      */
     private void serializePeerInfo(BinaryOutputStream stream, PeerInfo peer) {
-        if ((peer == thisPeer || peer.connected) && peer.peerType != TCP.PeerType.CLIENT_ONLY) {
+        if ((peer == thisPeer || peer.isConnected()) && peer.peerType != TCP.PeerType.CLIENT_ONLY) {
             stream.writeBoolean(true);
 
@@ -666 +672 @@
 
                     for (PeerInfo info : otherPeers) {
-                        if ((!info.connected) && (!info.connecting) && (info.peerType != PeerType.CLIENT_ONLY)) {
+                        if ((!info.isConnected()) && (!info.connecting) && (info.peerType != PeerType.CLIENT_ONLY)) {
                             new PeerConnectorThread(info).start();
                         }
@@ -689 +695 @@
             this.connectTo = connectTo;
             modelNode = new ModelNode("Looking for " + TCP.formatInetSocketAddress(connectTo) + "...");
-            connectionElement.getModelHandler().addNode(modelRootNode, modelNode);
+            BufferedModelChanges changes = new BufferedModelChanges();
+            changes.addNode(modelRootNode, modelNode);
+            connectionElement.getModelHandler().applyModelChanges(changes);
         }
 
@@ -698 +706 @@
                 stopThread();
                 if (modelRootNode != null) {
-                    connectionElement.getModelHandler().removeNode(modelNode);
+                    BufferedModelChanges changes = new BufferedModelChanges();
+                    changes.removeNode(modelNode);
+                    connectionElement.getModelHandler().applyModelChanges(changes);
                 }
                 if (peerInfo != null) {
@@ -710 +720 @@
             try {
                 socket.connect(connectTo);
-                connection1 = new TCPConnection(TCPPeer.this, socket, TCP.BULK_DATA, false, modelNode);
+                connection1 = new TCPConnection(TCPPeer.this, socket, TCP.PRIMARY_CONNECTION | TCP.BULK_DATA, false, modelNode);
                 //new TCPConnection(TCPPeer.this, socket, TCP.MANAGEMENT_DATA | TCP.EXPRESS_DATA, true, modelNode);
                 socket = new Socket();
                 socket.connect(connectTo);
                 //new TCPConnection(TCPPeer.this, socket, TCP.BULK_DATA, false, modelNode);
-                connection2 = new TCPConnection(TCPPeer.this, socket, TCP.MANAGEMENT_DATA | TCP.EXPRESS_DATA, true, modelNode);
+                connection2 = new TCPConnection(TCPPeer.this, socket, TCP.EXPRESS_DATA, true, modelNode);
                 peerInfo = findPeerInfoFor(connectTo);
                 peerInfo.remotePart.initAndCheckForAdminPort(modelNode);
@@ -733 +743 @@
                     peerInfo.remotePart.deleteAllChildren();
                 }
-                connectionElement.getModelHandler().changeNodeName(modelNode, "Looking for " + TCP.formatInetSocketAddress(connectTo) + "...");
-                //e.printStackTrace();
+                BufferedModelChanges changes = new BufferedModelChanges();
+                changes.changeNodeName(modelNode, "Looking for " + TCP.formatInetSocketAddress(connectTo) + "...");
+                connectionElement.getModelHandler().applyModelChanges(changes);
+                e.printStackTrace();
                 socket.close();
                 // simply try again
@@ -756 +768 @@
             peer.connecting = true;
             modelNode = new ModelNode("Looking for " + peer.uuid.toString() + "...");
-            connectionElement.getModelHandler().addNode(modelRootNode, modelNode);
+            BufferedModelChanges changes = new BufferedModelChanges();
+            changes.addNode(modelRootNode, modelNode);
+            connectionElement.getModelHandler().applyModelChanges(changes);
         }
 
         @Override
         public void mainLoopCallback() throws Exception {
-            if ((!activelyConnect) || peer.connected) {
+            if ((!activelyConnect) || peer.isConnected()) {
                 stopThread();
                 if (modelRootNode != null) {
-                    connectionElement.getModelHandler().removeNode(modelNode);
+                    BufferedModelChanges changes = new BufferedModelChanges();
+                    changes.removeNode(modelNode);
+                    connectionElement.getModelHandler().applyModelChanges(changes);
                 }
                 peer.connecting = false;
@@ -777 +793 @@
                 TCPConnection connection1 = null, connection2 = null;
                 try {
-                    if (peer.remotePart == null || peer.remotePart.bulkConnection == null) {
+                    RemotePart remotePart = peer.remotePart;
+                    if (remotePart != null && (!remotePart.isConnected())) {
+                        continue; // disconnecting - wait
+                    }
+
+                    if (remotePart == null || remotePart.getPrimaryConnection() == null) {
                         socket.connect(socketAddress);
                         //new TCPConnection(TCPPeer.this, socket, TCP.MANAGEMENT_DATA | TCP.EXPRESS_DATA, false, modelNode);
-                        connection1 = new TCPConnection(TCPPeer.this, socket, TCP.BULK_DATA, false, modelNode);
-                    }
-                    if (peer.remotePart == null || peer.remotePart.managementConnection == null) {
+                        connection1 = new TCPConnection(TCPPeer.this, socket, TCP.PRIMARY_CONNECTION | TCP.BULK_DATA, false, modelNode);
+                    }
+                    if (peer.remotePart == null || peer.remotePart.getExpressConnection() == null) {
                         socket = new Socket();
                         socket.connect(socketAddress);
                         //new TCPConnection(TCPPeer.this, socket, TCP.BULK_DATA, false, modelNode);
-                        connection2 = new TCPConnection(TCPPeer.this, socket, TCP.MANAGEMENT_DATA | TCP.EXPRESS_DATA, false, modelNode);
+                        connection2 = new TCPConnection(TCPPeer.this, socket, TCP.EXPRESS_DATA, false, modelNode);
                     }
                     peer.remotePart.initAndCheckForAdminPort(modelNode);
@@ -816 +837 @@
                         peer.remotePart.deleteAllChildren();
                     }
-                    connectionElement.getModelHandler().changeNodeName(modelNode, "Looking for " + peer.uuid.toString() + "...");
+                    BufferedModelChanges changes = new BufferedModelChanges();
+                    changes.changeNodeName(modelNode, "Looking for " + peer.uuid.toString() + "...");
+                    connectionElement.getModelHandler().applyModelChanges(changes);
                     socket.close();
                     // simply try again