class TransportWriteThrottle implements TransportThrottle {

    @Override
    public void acquire( Channel channel ) throws TransportThrottleException
    {
        // if this channel's max lock duration is already exceeded, we'll allow the protocol to
        // (at least) try to communicate the error to the client before aborting the connection
        if ( !isDurationAlreadyExceeded( channel ) )
        {
            ThrottleLock lock = channel.attr( LOCK_KEY ).get();

            long startTimeMillis = 0;
            while ( channel.isOpen() && !channel.isWritable() )
            {
                if ( maxLockDuration > 0 )
                {
                    long currentTimeMillis = clock.millis();
                    if ( startTimeMillis == 0 )
                    {
                        startTimeMillis = currentTimeMillis;
                    }
                    else
                    {
                        if ( currentTimeMillis - startTimeMillis > maxLockDuration )
                        {
                            setDurationExceeded( channel );

                            throw new TransportThrottleException( String.format(
                                    "Bolt connection [%s] will be closed because the client did not consume outgoing buffers for %s which is not expected.",
                                    channel.remoteAddress(), DurationFormatUtils.formatDurationHMS( maxLockDuration ) ) );
                        }
                    }
                }

                try
                {
                    lock.lock( channel, 1000 );
                }
                catch ( InterruptedException ex )
                {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException( ex );
                }
            }
        }
    }

    public  TransportWriteThrottle( int lowWaterMark, int highWaterMark, Clock clock, Duration maxLockDuration );
    public  TransportWriteThrottle( int lowWaterMark, int highWaterMark, Clock clock, Duration maxLockDuration, Supplier<ThrottleLock> lockSupplier );

}