@Test( timeout = 60_000L ) public void performanceTest() throws Exception { final long[] staredAt = {0}; final long[] finishedAt = {0}; final AtomicInteger calls = new AtomicInteger(); final int totalCalls = 1_000_000; EventHandler<Event> handler = new EventHandler<Event>() { @Override public void handleEvent( @Nonnull Event event ) { if ( staredAt[0] == 0 ) { staredAt[0] = System.currentTimeMillis(); } if ( calls.incrementAndGet() == totalCalls ) { finishedAt[0] = System.currentTimeMillis(); } } }; final ByteBuffer event = ByteBuffer.allocate( 192 ); event.put( ":maven-surefire-event:".getBytes( UTF_8 ) ); event.put( (byte) 14 ); event.put( ":std-out-stream:".getBytes( UTF_8 ) ); event.put( (byte) 10 ); event.put( ":normal-run:".getBytes( UTF_8 ) ); event.put( (byte) 5 ); event.put( ":UTF-8:".getBytes( UTF_8 ) ); event.putInt( 100 ); event.put( ":0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789:" .getBytes( UTF_8 ) ); event.flip(); byte[] frame = copyOfRange( event.array(), event.arrayOffset(), event.arrayOffset() + event.remaining() ); ReadableByteChannel channel = new Channel( frame, 1024 ) { private int countRounds; @Override public int read( ByteBuffer dst ) { int length = super.read( dst ); if ( length == -1 && countRounds < totalCalls ) { i = 0; length = super.read( dst ); countRounds++; } return length; } }; EventConsumerThread thread = new EventConsumerThread( "t", channel, handler, new CountdownCloseable( new MockCloseable(), 1 ), new MockForkNodeArguments() ); TimeUnit.SECONDS.sleep( 2 ); System.gc(); TimeUnit.SECONDS.sleep( 5 ); System.out.println( "Staring the event thread..." ); thread.start(); thread.join(); long execTime = finishedAt[0] - staredAt[0]; System.out.println( execTime ); assertThat( execTime ) .describedAs( "The performance test should assert 1.0s of read time. " + "The limit 3.6s guarantees that the read time does not exceed this limit on overloaded CPU." ) .isPositive() .isLessThanOrEqualTo( 3_600L ); }
protected StreamReadStatus read( Memento memento, int recommendedCount ) throws IOException { ByteBuffer buffer = memento.bb; if ( buffer.remaining() >= recommendedCount && buffer.position() != 0 ) { return OVERFLOW; } else { if ( buffer.position() != 0 && recommendedCount > buffer.capacity() - buffer.position() ) { buffer.compact().flip(); memento.line.positionByteBuffer = 0; } int mark = buffer.position(); buffer.position( buffer.limit() ); buffer.limit( buffer.capacity() ); boolean isEnd = false; while ( !isEnd && buffer.position() - mark < recommendedCount && buffer.position() != buffer.limit() ) { isEnd = channel.read( buffer ) == -1; } buffer.limit( buffer.position() ); buffer.position( mark ); int readBytes = buffer.remaining(); if ( isEnd && readBytes < recommendedCount ) { throw new EOFException(); } else { return readBytes >= recommendedCount ? OVERFLOW : UNDERFLOW; } } }
EventConsumerThread extends CloseableDaemonThread { protected StreamReadStatus read( Memento memento, int recommendedCount ) throws IOException { ByteBuffer buffer = memento.bb; if ( buffer.remaining() >= recommendedCount && buffer.position() != 0 ) { return OVERFLOW; } else { if ( buffer.position() != 0 && recommendedCount > buffer.capacity() - buffer.position() ) { buffer.compact().flip(); memento.line.positionByteBuffer = 0; } int mark = buffer.position(); buffer.position( buffer.limit() ); buffer.limit( buffer.capacity() ); boolean isEnd = false; while ( !isEnd && buffer.position() - mark < recommendedCount && buffer.position() != buffer.limit() ) { isEnd = channel.read( buffer ) == -1; } buffer.limit( buffer.position() ); buffer.position( mark ); int readBytes = buffer.remaining(); if ( isEnd && readBytes < recommendedCount ) { throw new EOFException(); } else { return readBytes >= recommendedCount ? OVERFLOW : UNDERFLOW; } } } }
EventConsumerThread extends CloseableDaemonThread { protected StreamReadStatus read( Memento memento, int recommendedCount ) throws IOException { ByteBuffer buffer = memento.bb; if ( buffer.remaining() >= recommendedCount && buffer.position() != 0 ) { return OVERFLOW; } else { if ( buffer.position() != 0 && recommendedCount > buffer.capacity() - buffer.position() ) { buffer.compact().flip(); memento.line.positionByteBuffer = 0; } int mark = buffer.position(); buffer.position( buffer.limit() ); buffer.limit( buffer.capacity() ); boolean isEnd = false; while ( !isEnd && buffer.position() - mark < recommendedCount && buffer.position() != buffer.limit() ) { isEnd = channel.read( buffer ) == -1; } buffer.limit( buffer.position() ); buffer.position( mark ); int readBytes = buffer.remaining(); if ( isEnd && readBytes < recommendedCount ) { throw new EOFException(); } else { return readBytes >= recommendedCount ? OVERFLOW : UNDERFLOW; } } } EventConsumerThread( @Nonnull String threadName,
                                @Nonnull ReadableByteChannel channel,
                                @Nonnull EventHandler<Event> eventHandler,
                                @Nonnull CountdownCloseable countdownCloseable,
                                @Nonnull ForkNodeArguments arguments ); }
EventConsumerThread extends CloseableDaemonThread { protected StreamReadStatus read( Memento memento, int recommendedCount ) throws IOException { ByteBuffer buffer = memento.bb; if ( buffer.remaining() >= recommendedCount && buffer.position() != 0 ) { return OVERFLOW; } else { if ( buffer.position() != 0 && recommendedCount > buffer.capacity() - buffer.position() ) { buffer.compact().flip(); memento.line.positionByteBuffer = 0; } int mark = buffer.position(); buffer.position( buffer.limit() ); buffer.limit( buffer.capacity() ); boolean isEnd = false; while ( !isEnd && buffer.position() - mark < recommendedCount && buffer.position() != buffer.limit() ) { isEnd = channel.read( buffer ) == -1; } buffer.limit( buffer.position() ); buffer.position( mark ); int readBytes = buffer.remaining(); if ( isEnd && readBytes < recommendedCount ) { throw new EOFException(); } else { return readBytes >= recommendedCount ? OVERFLOW : UNDERFLOW; } } } EventConsumerThread( @Nonnull String threadName,
                                @Nonnull ReadableByteChannel channel,
                                @Nonnull EventHandler<Event> eventHandler,
                                @Nonnull CountdownCloseable countdownCloseable,
                                @Nonnull ForkNodeArguments arguments ); @Override void run(); @Override void disable(); @Override void close(); }
EventConsumerThread extends CloseableDaemonThread { protected StreamReadStatus read( Memento memento, int recommendedCount ) throws IOException { ByteBuffer buffer = memento.bb; if ( buffer.remaining() >= recommendedCount && buffer.position() != 0 ) { return OVERFLOW; } else { if ( buffer.position() != 0 && recommendedCount > buffer.capacity() - buffer.position() ) { buffer.compact().flip(); memento.line.positionByteBuffer = 0; } int mark = buffer.position(); buffer.position( buffer.limit() ); buffer.limit( buffer.capacity() ); boolean isEnd = false; while ( !isEnd && buffer.position() - mark < recommendedCount && buffer.position() != buffer.limit() ) { isEnd = channel.read( buffer ) == -1; } buffer.limit( buffer.position() ); buffer.position( mark ); int readBytes = buffer.remaining(); if ( isEnd && readBytes < recommendedCount ) { throw new EOFException(); } else { return readBytes >= recommendedCount ? OVERFLOW : UNDERFLOW; } } } EventConsumerThread( @Nonnull String threadName,
                                @Nonnull ReadableByteChannel channel,
                                @Nonnull EventHandler<Event> eventHandler,
                                @Nonnull CountdownCloseable countdownCloseable,
                                @Nonnull ForkNodeArguments arguments ); @Override void run(); @Override void disable(); @Override void close();  }