diff --git a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
index 5441dec..33e4f87 100644
--- a/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
+++ b/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
@@ -1554,15 +1554,14 @@
     @Override
     public long tick(long now)
     {
-        long timeout = 0;
+        long deadline = 0;
 
         if (_localIdleTimeout > 0) {
             if (_localIdleDeadline == 0 || _lastBytesInput != _bytesInput) {
-                _localIdleDeadline = now + _localIdleTimeout;
+                _localIdleDeadline = computeDeadline(now, _localIdleTimeout);
                 _lastBytesInput = _bytesInput;
-            } else if (_localIdleDeadline <= now) {
-                _localIdleDeadline = now + _localIdleTimeout;
-
+            } else if (_localIdleDeadline - now <= 0) {
+                _localIdleDeadline = computeDeadline(now, _localIdleTimeout);
                 if (_connectionEndpoint != null &&
                     _connectionEndpoint.getLocalState() != EndpointState.CLOSED) {
                     ErrorCondition condition =
@@ -1588,24 +1587,41 @@
                     close_tail();
                 }
             }
-            timeout = _localIdleDeadline;
+            deadline = _localIdleDeadline;
         }
 
         if (_remoteIdleTimeout != 0 && !_isCloseSent) {
             if (_remoteIdleDeadline == 0 || _lastBytesOutput != _bytesOutput) {
-                _remoteIdleDeadline = now + (_remoteIdleTimeout / 2);
+                _remoteIdleDeadline = computeDeadline(now, _remoteIdleTimeout / 2);
                 _lastBytesOutput = _bytesOutput;
-            } else if (_remoteIdleDeadline <= now) {
-                _remoteIdleDeadline = now + (_remoteIdleTimeout / 2);
+            } else if (_remoteIdleDeadline - now <= 0) {
+                _remoteIdleDeadline = computeDeadline(now, _remoteIdleTimeout / 2);
                 if (pending() == 0) {
                     writeFrame(0, null, null, null);
                     _lastBytesOutput += pending();
                 }
             }
-            timeout = Math.min(timeout == 0 ? _remoteIdleDeadline : timeout, _remoteIdleDeadline);
+
+            if(deadline == 0) {
+                deadline = _remoteIdleDeadline;
+            } else {
+                if(_remoteIdleDeadline - _localIdleDeadline <= 0) {
+                    deadline = _remoteIdleDeadline;
+                } else {
+                    deadline = _localIdleDeadline;
+                }
+            }
         }
 
-        return timeout;
+        return deadline;
+    }
+
+    private long computeDeadline(long now, long timeout) {
+        long deadline = now + timeout;
+
+        // We use 0 to signal not-initialised and/or no-timeout, so in the
+        // unlikely event thats to be the actual deadline, return 1 instead
+        return deadline != 0 ? deadline : 1;
     }
 
     @Override
diff --git a/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java b/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java
index f6ca12c..80ad20c 100644
--- a/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java
+++ b/proton-j/src/test/java/org/apache/qpid/proton/engine/impl/TransportImplTest.java
@@ -362,15 +362,7 @@
 
         // Protocol header + empty frame
         ByteBuffer data = ByteBuffer.wrap(new byte[] {'A', 'M', 'Q', 'P', 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00});
-        while (data.remaining() > 0)
-        {
-            int origLimit = data.limit();
-            int amount = Math.min(transport.tail().remaining(), data.remaining());
-            data.limit(data.position() + amount);
-            transport.tail().put(data);
-            data.limit(origLimit);
-            transport.process();
-        }
+        processInput(transport, data);
         framesWrittenBeforeTick = transport.writes.size();
         deadline = transport.tick(2000);
         assertEquals("Reading data data resets the deadline",  6000, deadline);
@@ -1667,4 +1659,488 @@
 
         assertEquals("Unexpected value : " + getFrameTypesWritten(transport), 4567, transport.getRemoteMaxFrameSize());
     }
+
+    @Test
+    public void testTickWithZeroIdleTimeoutsGivesZeroDeadline()
+    {
+        doTickWithNoIdleTimeoutGivesZeroDeadlineTestImpl(true);
+    }
+
+    @Test
+    public void testTickWithNullIdleTimeoutsGivesZeroDeadline()
+    {
+        doTickWithNoIdleTimeoutGivesZeroDeadlineTestImpl(false);
+    }
+
+    private void doTickWithNoIdleTimeoutGivesZeroDeadlineTestImpl(boolean useZero) {
+        MockTransportImpl transport = new MockTransportImpl();
+        Connection connection = Proton.connection();
+        transport.bind(connection);
+
+        connection.open();
+        while(transport.pending() > 0) {
+            transport.pop(transport.head().remaining());
+        }
+
+        assertEquals("should have written data", 1, transport.writes.size());
+        FrameBody sentOpenFrame = transport.writes.get(0);
+        assertNotNull("should have written a non-empty frame", sentOpenFrame);
+        assertTrue("should have written an open frame", sentOpenFrame instanceof Open);
+        assertNull("should not have had an idletimeout value", ((Open)sentOpenFrame).getIdleTimeOut());
+
+        // Handle the peer transmitting their open with null/zero timeout.
+        Open open = new Open();
+        if(useZero) {
+            open.setIdleTimeOut(UnsignedInteger.ZERO);
+        } else {
+            open.setIdleTimeOut(null);
+        }
+        TransportFrame openFrame = new TransportFrame(CHANNEL_ID, open, null);
+        transport.handleFrame(openFrame);
+        pumpMockTransport(transport);
+
+        long deadline = transport.tick(0);
+        assertEquals("Unexpected deadline returned", 0, deadline);
+
+        deadline = transport.tick(10);
+        assertEquals("Unexpected deadline returned", 0, deadline);
+    }
+
+    @Test
+    public void testTickWithBothTimeouts()
+    {
+        // all-positive
+        doTickWithBothTimeoutsTestImpl(true, 5000, 2000, 10000, 12000, 14000, 15000);
+        doTickWithBothTimeoutsTestImpl(false, 5000, 2000, 10000, 12000, 14000, 15000);
+
+        // all-negative
+        doTickWithBothTimeoutsTestImpl(true, 10000, 4000, -100000, -96000, -92000, -90000);
+        doTickWithBothTimeoutsTestImpl(false, 10000, 4000, -100000, -96000, -92000, -90000);
+
+        // negative to positive missing 0
+        doTickWithBothTimeoutsTestImpl(true, 500, 200, -450, -250, -50, 50);
+        doTickWithBothTimeoutsTestImpl(false, 500, 200, -450, -250, -50, 50);
+
+        // negative to positive striking 0 with local deadline
+        doTickWithBothTimeoutsTestImpl(true, 500, 200, -500, -300, -100, 1);
+        doTickWithBothTimeoutsTestImpl(false, 500, 200, -500, -300, -100, 1);
+
+        // negative to positive striking 0 with remote deadline
+        doTickWithBothTimeoutsTestImpl(true, 500, 200, -200, 1, 201, 300);
+        doTickWithBothTimeoutsTestImpl(false, 500, 200, -200, 1, 201, 300);
+    }
+
+    private void doTickWithBothTimeoutsTestImpl(boolean allowLocalTimeout, int localTimeout, int remoteTimeoutHalf, long tick1,
+                                                long expectedDeadline1, long expectedDeadline2, long expectedDeadline3)
+    {
+        MockTransportImpl transport = new MockTransportImpl();
+        Connection connection = Proton.connection();
+        transport.bind(connection);
+
+        // Set our local idleTimeout
+        transport.setIdleTimeout(localTimeout);
+
+        connection.open();
+        pumpMockTransport(transport);
+
+        assertEquals("should have written data", 1, transport.writes.size());
+        assertNotNull("should have written a non-empty frame", transport.writes.get(0));
+        assertTrue("should have written an open frame", transport.writes.get(0) instanceof Open);
+
+        // Handle the peer transmitting [half] their timeout. We half it on receipt to avoid spurious timeouts
+        // if they not have transmitted half their actual timeout, as the AMQP spec only says they SHOULD do that.
+        Open open = new Open();
+        open.setIdleTimeOut(new UnsignedInteger(remoteTimeoutHalf * 2));
+        TransportFrame openFrame = new TransportFrame(CHANNEL_ID, open, null);
+        transport.handleFrame(openFrame);
+        pumpMockTransport(transport);
+
+        long deadline = transport.tick(tick1);
+        assertEquals("Unexpected deadline returned", expectedDeadline1, deadline);
+
+        // Wait for less time than the deadline with no data - get the same value
+        long interimTick = tick1 + 10;
+        assertTrue (interimTick < expectedDeadline1);
+        assertEquals("When the deadline hasn't been reached tick() should return the previous deadline",  expectedDeadline1, transport.tick(interimTick));
+        assertEquals("When the deadline hasn't been reached tick() shouldn't write data", 1, transport.writes.size());
+
+        deadline = transport.tick(expectedDeadline1);
+        assertEquals("When the deadline has been reached expected a new remote deadline to be returned", expectedDeadline2, deadline);
+        assertEquals("tick() should have written data", 2, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(1));
+
+        pumpMockTransport(transport);
+
+        deadline = transport.tick(expectedDeadline2);
+        assertEquals("When the deadline has been reached expected a new local deadline to be returned", expectedDeadline3, deadline);
+        assertEquals("tick() should have written data", 3, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(2));
+
+        pumpMockTransport(transport);
+
+        if(allowLocalTimeout) {
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+            transport.tick(expectedDeadline3); // Wait for the deadline, but don't receive traffic, allow local timeout to expire
+            assertEquals("Calling tick() after the deadline should result in the connection being closed", EndpointState.CLOSED, connection.getLocalState());
+            assertEquals("tick() should have written data", 4, transport.writes.size());
+            assertNotNull("should have written a non-empty frame", transport.writes.get(3));
+            assertTrue("should have written a close frame", transport.writes.get(3) instanceof Close);
+        } else {
+            // Receive Empty frame to satisfy local deadline
+            processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+            deadline = transport.tick(expectedDeadline3);
+            assertEquals("Receiving data should have reset the deadline (to the next remote one)",  expectedDeadline2 + (remoteTimeoutHalf), deadline);
+            assertEquals("tick() shouldn't have written data", 3, transport.writes.size());
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+        }
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsLocalThenRemote()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsLocalThenRemoteTestImpl(false);
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsLocalThenRemoteWithLocalTimeout()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsLocalThenRemoteTestImpl(true);
+    }
+
+    private void doTickWithNanoTimeDerivedValueWhichWrapsLocalThenRemoteTestImpl(boolean allowLocalTimeout) {
+        int localTimeout = 5000;
+        int remoteTimeoutHalf = 2000;
+        assertTrue(remoteTimeoutHalf < localTimeout);
+
+        long offset = 2500;
+        assertTrue(offset < localTimeout);
+        assertTrue(offset > remoteTimeoutHalf);
+
+        MockTransportImpl transport = new MockTransportImpl();
+        Connection connection = Proton.connection();
+        transport.bind(connection);
+
+        // Set our local idleTimeout
+        transport.setIdleTimeout(localTimeout);
+
+        connection.open();
+        pumpMockTransport(transport);
+
+        assertEquals("should have written data", 1, transport.writes.size());
+        assertNotNull("should have written a non-empty frame", transport.writes.get(0));
+        assertTrue("should have written an open frame", transport.writes.get(0) instanceof Open);
+
+        // Handle the peer transmitting [half] their timeout. We half it on receipt to avoid spurious timeouts
+        // if they not have transmitted half their actual timeout, as the AMQP spec only says they SHOULD do that.
+        Open open = new Open();
+        open.setIdleTimeOut(new UnsignedInteger(remoteTimeoutHalf * 2));
+        TransportFrame openFrame = new TransportFrame(CHANNEL_ID, open, null);
+        transport.handleFrame(openFrame);
+        pumpMockTransport(transport);
+
+        long deadline = transport.tick(Long.MAX_VALUE - offset);
+        assertEquals("Unexpected deadline returned", Long.MAX_VALUE - offset + remoteTimeoutHalf, deadline);
+
+        deadline = transport.tick(Long.MAX_VALUE - (offset - 100));    // Wait for less time than the deadline with no data - get the same value
+        assertEquals("When the deadline hasn't been reached tick() should return the previous deadline",  Long.MAX_VALUE -offset + remoteTimeoutHalf, deadline);
+        assertEquals("When the deadline hasn't been reached tick() shouldn't write data", 1, transport.writes.size());
+
+        deadline = transport.tick(Long.MAX_VALUE -offset + remoteTimeoutHalf); // Wait for the deadline - next deadline should be previous + remoteTimeoutHalf;
+        assertEquals("When the deadline has been reached expected a new remote deadline to be returned", Long.MIN_VALUE + (2* remoteTimeoutHalf) - offset -1, deadline);
+        assertEquals("tick() should have written data", 2, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(1));
+
+        pumpMockTransport(transport);
+
+        deadline = transport.tick(Long.MIN_VALUE + (2* remoteTimeoutHalf) - offset -1); // Wait for the deadline - next deadline should be orig + localTimeout;
+        assertEquals("When the deadline has been reached expected a new local deadline to be returned", Long.MIN_VALUE + (localTimeout - offset) -1, deadline);
+        assertEquals("tick() should have written data", 3, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(2));
+
+        pumpMockTransport(transport);
+
+        if(allowLocalTimeout) {
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+            transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1); // Wait for the deadline, but don't receive traffic, allow local timeout to expire
+            assertEquals("Calling tick() after the deadline should result in the connection being closed", EndpointState.CLOSED, connection.getLocalState());
+            assertEquals("tick() should have written data", 4, transport.writes.size());
+            assertNotNull("should have written a non-empty frame", transport.writes.get(3));
+            assertTrue("should have written a close frame", transport.writes.get(3) instanceof Close);
+        } else {
+            // Receive Empty frame to satisfy local deadline
+            processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+            deadline = transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1); // Wait for the deadline - next deadline should be orig + 3*remoteTimeoutHalf;
+            assertEquals("Receiving data should have reset the deadline (to the remote one)",  Long.MIN_VALUE + (3* remoteTimeoutHalf) - offset -1, deadline);
+            assertEquals("tick() shouldn't have written data", 3, transport.writes.size());
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+        }
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsRemoteThenLocal()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsRemoteThenLocalTestImpl(false);
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsRemoteThenLocalWithLocalTimeout()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsRemoteThenLocalTestImpl(true);
+    }
+
+    private void doTickWithNanoTimeDerivedValueWhichWrapsRemoteThenLocalTestImpl(boolean allowLocalTimeout) {
+        int localTimeout = 2000;
+        int remoteTimeoutHalf = 5000;
+        assertTrue(localTimeout < remoteTimeoutHalf);
+
+        long offset = 2500;
+        assertTrue(offset > localTimeout);
+        assertTrue(offset < remoteTimeoutHalf);
+
+        MockTransportImpl transport = new MockTransportImpl();
+        Connection connection = Proton.connection();
+        transport.bind(connection);
+
+        // Set our local idleTimeout
+        transport.setIdleTimeout(localTimeout);
+
+        connection.open();
+        pumpMockTransport(transport);
+
+        assertEquals("should have written data", 1, transport.writes.size());
+        assertNotNull("should have written a non-empty frame", transport.writes.get(0));
+        assertTrue("should have written an open frame", transport.writes.get(0) instanceof Open);
+
+        // Receive Protocol header
+        processInput(transport, ByteBuffer.wrap(new byte[] {'A', 'M', 'Q', 'P', 0x00, 0x01, 0x00, 0x00}));
+
+        // Handle the peer transmitting [half] their timeout. We half it on receipt to avoid spurious timeouts
+        // if they not have transmitted half their actual timeout, as the AMQP spec only says they SHOULD do that.
+        Open open = new Open();
+        open.setIdleTimeOut(new UnsignedInteger(remoteTimeoutHalf * 2));
+        TransportFrame openFrame = new TransportFrame(CHANNEL_ID, open, null);
+        transport.handleFrame(openFrame);
+        pumpMockTransport(transport);
+
+        long deadline = transport.tick(Long.MAX_VALUE - offset);
+        assertEquals("Unexpected deadline returned",  Long.MAX_VALUE - offset + localTimeout, deadline);
+
+        deadline = transport.tick(Long.MAX_VALUE - (offset - 100));    // Wait for less time than the deadline with no data - get the same value
+        assertEquals("When the deadline hasn't been reached tick() should return the previous deadline",  Long.MAX_VALUE - offset + localTimeout, deadline);
+        assertEquals("tick() shouldn't have written data", 1, transport.writes.size());
+
+        // Receive Empty frame to satisfy local deadline
+        processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+        deadline = transport.tick(Long.MAX_VALUE - offset + localTimeout); // Wait for the deadline - next deadline should be orig + 2* localTimeout;
+        assertEquals("When the deadline has been reached expected a new local deadline to be returned", Long.MIN_VALUE + (localTimeout - offset) -1 + localTimeout, deadline);
+        assertEquals("tick() should not have written data", 1, transport.writes.size());
+
+        pumpMockTransport(transport);
+
+        if(allowLocalTimeout) {
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+            transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1 + localTimeout); // Wait for the deadline, but don't receive traffic, allow local timeout to expire
+            assertEquals("Calling tick() after the deadline should result in the connection being closed", EndpointState.CLOSED, connection.getLocalState());
+            assertEquals("tick() should have written data", 2, transport.writes.size());
+            assertNotNull("should have written a non-empty frame", transport.writes.get(1));
+            assertTrue("should have written a close frame", transport.writes.get(1) instanceof Close);
+        } else {
+            // Receive Empty frame to satisfy local deadline
+            processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+            deadline = transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1 + localTimeout); // Wait for the deadline - next deadline should be orig + remoteTimeoutHalf;
+            assertEquals("Receiving data should have reset the deadline (to the remote one)",  Long.MIN_VALUE + remoteTimeoutHalf - offset -1, deadline);
+            assertEquals("tick() shouldn't have written data", 1, transport.writes.size());
+
+            deadline = transport.tick(Long.MIN_VALUE + remoteTimeoutHalf - offset -1); // Wait for the deadline - next deadline should be orig + 3* localTimeout;
+            assertEquals("When the deadline has been reached expected a new local deadline to be returned", Long.MIN_VALUE + (3* localTimeout) - offset -1, deadline);
+            assertEquals("tick() should have written data", 2, transport.writes.size());
+            assertEquals("tick() should have written an empty frame", null, transport.writes.get(1));
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+        }
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsBothRemoteFirst()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsBothRemoteFirstTestImpl(false);
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsBothRemoteFirstWithLocalTimeout()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsBothRemoteFirstTestImpl(true);
+    }
+
+    private void doTickWithNanoTimeDerivedValueWhichWrapsBothRemoteFirstTestImpl(boolean allowLocalTimeout) {
+        int localTimeout = 2000;
+        int remoteTimeoutHalf = 2500;
+        assertTrue(localTimeout < remoteTimeoutHalf);
+
+        long offset = 500;
+        assertTrue(offset < localTimeout);
+
+        MockTransportImpl transport = new MockTransportImpl();
+        Connection connection = Proton.connection();
+        transport.bind(connection);
+
+        // Set our local idleTimeout
+        transport.setIdleTimeout(localTimeout);
+
+        connection.open();
+        pumpMockTransport(transport);
+
+        assertEquals("should have written data", 1, transport.writes.size());
+        assertNotNull("should have written a non-empty frame", transport.writes.get(0));
+        assertTrue("should have written an open frame", transport.writes.get(0) instanceof Open);
+
+        // Receive Protocol header
+        processInput(transport, ByteBuffer.wrap(new byte[] {'A', 'M', 'Q', 'P', 0x00, 0x01, 0x00, 0x00}));
+
+        // Handle the peer transmitting [half] their timeout. We half it on receipt to avoid spurious timeouts
+        // if they not have transmitted half their actual timeout, as the AMQP spec only says they SHOULD do that.
+        Open open = new Open();
+        open.setIdleTimeOut(new UnsignedInteger(remoteTimeoutHalf * 2));
+        TransportFrame openFrame = new TransportFrame(CHANNEL_ID, open, null);
+        transport.handleFrame(openFrame);
+        pumpMockTransport(transport);
+
+        long deadline = transport.tick(Long.MAX_VALUE - offset);
+        assertEquals("Unexpected deadline returned",  Long.MIN_VALUE + (localTimeout - offset) -1, deadline);
+
+        deadline = transport.tick(Long.MAX_VALUE - (offset - 100));    // Wait for less time than the deadline with no data - get the same value
+        assertEquals("When the deadline hasn't been reached tick() should return the previous deadline",  Long.MIN_VALUE + (localTimeout - offset) -1, deadline);
+        assertEquals("tick() shouldn't have written data", 1, transport.writes.size());
+
+        // Receive Empty frame to satisfy local deadline
+        processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+        deadline = transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1); // Wait for the deadline - next deadline should be orig + remoteTimeoutHalf;
+        assertEquals("When the deadline has been reached expected a new remote deadline to be returned", Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1, deadline);
+        assertEquals("When the deadline hasn't been reached tick() shouldn't write data", 1, transport.writes.size());
+
+        deadline = transport.tick(Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1); // Wait for the deadline - next deadline should be orig + 2* localTimeout;
+        assertEquals("When the deadline has been reached expected a new local deadline to be returned", Long.MIN_VALUE + (localTimeout - offset) -1 + localTimeout, deadline);
+        assertEquals("tick() should have written data", 2, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(1));
+
+        pumpMockTransport(transport);
+
+        if(allowLocalTimeout) {
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+            transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1 + localTimeout); // Wait for the deadline, but don't receive traffic, allow local timeout to expire
+            assertEquals("Calling tick() after the deadline should result in the connection being closed", EndpointState.CLOSED, connection.getLocalState());
+            assertEquals("tick() should have written data", 3, transport.writes.size());
+            assertNotNull("should have written a non-empty frame", transport.writes.get(2));
+            assertTrue("should have written a close frame", transport.writes.get(2) instanceof Close);
+        } else {
+            // Receive Empty frame to satisfy local deadline
+            processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+            deadline = transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1 + localTimeout); // Wait for the deadline - next deadline should be orig + 2*remoteTimeoutHalf;
+            assertEquals("Receiving data should have reset the deadline (to the remote one)",  Long.MIN_VALUE + (2* remoteTimeoutHalf) - offset -1, deadline);
+            assertEquals("tick() shouldn't have written data", 2, transport.writes.size());
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+        }
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsBothLocalFirst()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsBothLocalFirstTestImpl(false);
+    }
+
+    @Test
+    public void testTickWithNanoTimeDerivedValueWhichWrapsBothLocalFirstWithLocalTimeout()
+    {
+        doTickWithNanoTimeDerivedValueWhichWrapsBothLocalFirstTestImpl(true);
+    }
+
+    private void doTickWithNanoTimeDerivedValueWhichWrapsBothLocalFirstTestImpl(boolean allowLocalTimeout) {
+        int localTimeout = 5000;
+        int remoteTimeoutHalf = 2000;
+        assertTrue(remoteTimeoutHalf < localTimeout);
+
+        long offset = 500;
+        assertTrue(offset < remoteTimeoutHalf);
+
+        MockTransportImpl transport = new MockTransportImpl();
+        Connection connection = Proton.connection();
+        transport.bind(connection);
+
+        // Set our local idleTimeout
+        transport.setIdleTimeout(localTimeout);
+
+        connection.open();
+        while(transport.pending() > 0) {
+            transport.pop(transport.head().remaining());
+        }
+
+        assertEquals("should have written data", 1, transport.writes.size());
+        assertNotNull("should have written a non-empty frame", transport.writes.get(0));
+        assertTrue("should have written an open frame", transport.writes.get(0) instanceof Open);
+
+        // Handle the peer transmitting [half] their timeout. We half it on receipt to avoid spurious timeouts
+        // if they not have transmitted half their actual timeout, as the AMQP spec only says they SHOULD do that.
+        Open open = new Open();
+        open.setIdleTimeOut(new UnsignedInteger(remoteTimeoutHalf * 2));
+        TransportFrame openFrame = new TransportFrame(CHANNEL_ID, open, null);
+        transport.handleFrame(openFrame);
+        pumpMockTransport(transport);
+
+
+        long deadline = transport.tick(Long.MAX_VALUE - offset);
+        assertEquals("Unexpected deadline returned",  Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1, deadline);
+
+        deadline = transport.tick(Long.MAX_VALUE - (offset - 100));    // Wait for less time than the deadline with no data - get the same value
+        assertEquals("When the deadline hasn't been reached tick() should return the previous deadline",  Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1, deadline);
+        assertEquals("When the deadline hasn't been reached tick() shouldn't write data", 1, transport.writes.size());
+
+        deadline = transport.tick(Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1); // Wait for the deadline - next deadline should be previous + remoteTimeoutHalf;
+        assertEquals("When the deadline has been reached expected a new remote deadline to be returned", Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1 + remoteTimeoutHalf, deadline);
+        assertEquals("tick() should have written data", 2, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(1));
+
+        pumpMockTransport(transport);
+
+        deadline = transport.tick(Long.MIN_VALUE + (remoteTimeoutHalf - offset) -1 + remoteTimeoutHalf); // Wait for the deadline - next deadline should be orig + localTimeout;
+        assertEquals("When the deadline has been reached expected a new local deadline to be returned", Long.MIN_VALUE + (localTimeout - offset) -1, deadline);
+        assertEquals("tick() should have written data", 3, transport.writes.size());
+        assertEquals("tick() should have written an empty frame", null, transport.writes.get(2));
+
+        pumpMockTransport(transport);
+
+        if(allowLocalTimeout) {
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+            transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1); // Wait for the deadline, but don't receive traffic, allow local timeout to expire
+            assertEquals("Calling tick() after the deadline should result in the connection being closed", EndpointState.CLOSED, connection.getLocalState());
+            assertEquals("tick() should have written data", 4, transport.writes.size());
+            assertNotNull("should have written a non-empty frame", transport.writes.get(3));
+            assertTrue("should have written a close frame", transport.writes.get(3) instanceof Close);
+        } else {
+            // Receive Empty frame to satisfy local deadline
+            processInput(transport,  ByteBuffer.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00}));
+
+            deadline = transport.tick(Long.MIN_VALUE + (localTimeout - offset) -1); // Wait for the deadline - next deadline should be orig + 3*remoteTimeoutHalf;
+            assertEquals("Receiving data should have reset the deadline (to the remote one)",  Long.MIN_VALUE + (3* remoteTimeoutHalf) - offset -1, deadline);
+            assertEquals("tick() shouldn't have written data", 3, transport.writes.size());
+            assertEquals("Connection should be active", EndpointState.ACTIVE, connection.getLocalState());
+        }
+    }
+
+    private void processInput(MockTransportImpl transport, ByteBuffer data) {
+        while (data.remaining() > 0)
+        {
+            int origLimit = data.limit();
+            int amount = Math.min(transport.tail().remaining(), data.remaining());
+            data.limit(data.position() + amount);
+            transport.tail().put(data);
+            data.limit(origLimit);
+            transport.process();
+        }
+    }
+
 }