### Question:
BoltConnectionReadLimiter implements BoltConnectionQueueMonitor { @Override public synchronized void enqueued( BoltConnection to, Job job ) { queueSize += 1; checkLimitsOnEnqueue( to ); } BoltConnectionReadLimiter( LogService logService, int lowWatermark, int highWatermark ); @Override synchronized void enqueued( BoltConnection to, Job job ); @Override synchronized void drained( BoltConnection from, Collection<Job> batch );  }### Answer:
@Test public void shouldNotDisableAutoReadBelowHighWatermark() { BoltConnectionReadLimiter limiter = newLimiter( 1, 2 ); assertTrue( channel.config().isAutoRead() ); limiter.enqueued( connection, job ); assertTrue( channel.config().isAutoRead() ); verify( log, never() ).warn( anyString(), any(), any() ); }
@Test public void shouldDisableAutoReadWhenAtHighWatermark() { BoltConnectionReadLimiter limiter = newLimiter( 1, 2 ); assertTrue( channel.config().isAutoRead() ); limiter.enqueued( connection, job ); limiter.enqueued( connection, job ); limiter.enqueued( connection, job ); assertFalse( channel.config().isAutoRead() ); verify( log ).warn( contains( "disabled" ), eq( channel.remoteAddress() ), eq( 3 ) ); }
@Test public void shouldDisableAutoReadOnlyOnceWhenAboveHighWatermark() { BoltConnectionReadLimiter limiter = newLimiter( 1, 2 ); assertTrue( channel.config().isAutoRead() ); limiter.enqueued( connection, job ); limiter.enqueued( connection, job ); limiter.enqueued( connection, job ); limiter.enqueued( connection, job ); limiter.enqueued( connection, job ); assertFalse( channel.config().isAutoRead() ); verify( log, times( 1 ) ).warn( contains( "disabled" ), eq( channel.remoteAddress() ), eq( 3 ) ); }