How can I get progress rate when uploading big file? Get it when uploading not completed!

[yizhl]

I am trying to implement async file upload with progress ,get progress rate when I uploading a big file in order to show it to user..

I tried the method suggested in the docs. Using transfer listener. http://sonatype.github.com/async-http-client/transfer-listener.html

I implemented onBytesSent of TransferListener interface (just as test):

public void onBytesSent(ByteBuffer byteBuffer) {

     System.out.println("Total bytes sent - ");
     System.out.println(byteBuffer.capacity());
}

Then in another thread(because I don't want to block the app) I tried to do the following:

TransferCompletionHandler tl = new TransferCompletionHandler();
tl.addTransferListener(listener);

asyncHttpClient.preparePut(getFullUrl(fileWithPath))
      .setBody(new BodyGenerator() {
           public Body createBody() throws IOException {
                return new FileBodyWithOffset(file, offset);
           }
       })
       .addHeader(CONTENT_RANGE, new ContentRange(offset, localSize).toString())
       .execute(handler).get();

Everything is fine. File is uploaded correctly and very fast. But the issue is - I am getting messages from onBytesSent in TransferListener only AFTER the upload is finished. For exmaple the upload is completed in 10 minutes. And during that 10 minutes I get nothing. And only after that everything is printed on the console.

I can't figure out what is wrong with this code. I just tried to follow the docs.

I tried to execute the above code in the main thread and it didn't work either.

Maybe it is a wrong way to implement upload progress listener using this client?

[yizhl]

TransferCompletionHandler handler = new TransferCompletionHandler();
        handler.addTransferListener(new TransferListener() {
            @Override
            public void onRequestResponseCompleted() {}
            @Override
            public void onResponseHeadersReceived(FluentCaseInsensitiveStringsMap entries) {}
            /**
             * Invoked when the request bytes are starting to get send
            * */
            @Override
            public void onRequestHeadersSent(FluentCaseInsensitiveStringsMap entries) {
                //Connection [keep-alive]
                //Host [xxxx.com]
                //Accept [*/*]
                //User-Agent [AHC/1.0]
                System.out.println("request header sent");
            }
            @Override
            public void onBytesReceived(byte[] bytes) throws IOException {
            }
            /**
             * Invoked every time request's chunk are sent.
             *
             * @param amount
             * @param current
             * @param total
             */
            @Override
            public void <b>onBytesSent</b>(long amount, long current, long total) {
                    System.out.println("amount = " + amount+" current"+current+" total="+total);
//                float percent = current/amount;
//                System.setProperty("currentUpload", String.valueOf(percent));
            }

            @Override
            public void onThrowable(Throwable throwable) {
                String errorMsg = ExceptionUtil.exceptionChainToString(throwable);
                System.out.println("some error happend："+errorMsg);
                client.close();
            }
        });

Use TransferCompletionHandler or SimpleAHCTransferListener or similar handler,they all only show the progress after uploading finished, you can not get it when file is uploading!
I used latest version

 <dependency>
            <groupId>com.ning</groupId>
            <artifactId>async-http-client</artifactId>
            <version>1.9.0-BETA16</version>
</dependency>

[yizhl]

I tried to use SimpleAsyncHttpClient.post(BodyGenerator) to overide read method as below:

package net.async.http;

import com.ning.http.client.BodyGenerator;
import com.ning.http.client.RandomAccessBody;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.WritableByteChannel;

public class MyFileBodyGenerator  implements BodyGenerator {
    private final File file;
    private final long regionSeek;
    private final long regionLength;
    private FileBody fileBody;

    public MyFileBodyGenerator(File file) {
        if (file == null)
            throw new NullPointerException("file");
        this.file = file;
        this.regionLength = file.length();
        this.regionSeek = 0;
    }

    public MyFileBodyGenerator(File file, long regionSeek, long regionLength) {
        if (file == null)
            throw new NullPointerException("file");
        this.file = file;
        this.regionLength = regionLength;
        this.regionSeek = regionSeek;
    }

    public FileBody getFileBody() {
        return fileBody;
    }

    @Override
    public RandomAccessBody createBody()
            throws IOException {
        FileBody fileBody = new FileBody(file, regionSeek, regionLength);
        this.fileBody = fileBody;
        return fileBody;
    }

    public File getFile() {
        return file;
    }

    public long getRegionSeek() {
        return regionSeek;
    }

    public long getRegionLength() {
        return regionLength;
    }

    public static class FileBody
            implements RandomAccessBody {

        private final RandomAccessFile file;

        private final FileChannel channel;

        private final long length;
        public long uploadingPosition;  //current position

        public long getUploadingPosition() {
            return uploadingPosition;
        }

        public FileBody(File file)
                throws IOException {
            this.file = new RandomAccessFile(file, "r");
            channel = this.file.getChannel();
            length = file.length();
        }

        public FileBody(File file, long regionSeek, long regionLength)
                throws IOException {
            this.file = new RandomAccessFile(file, "r");
            channel = this.file.getChannel();
            length = regionLength;
            if (regionSeek > 0) {
                this.file.seek(regionSeek);
            }
        }

        @Override
        public long getContentLength() {
            return length;
        }

        // => override start
        @Override
        public long read(ByteBuffer buffer)
                throws IOException {
            this.uploadingPosition += buffer.capacity();
            System.out.println(uploadingPosition);   
            return channel.read(buffer);
        }

        @Override
        public long transferTo(long position, WritableByteChannel target)
                throws IOException {
            return channel.transferTo(position, length, target);
        }
        // <= override end
        @Override
        public void close() throws IOException {
            file.close();
        }
    }
}

From the source code,it's seems override read or transferTo method will able to get progress,but actually you can't!
The read method just read things to memory， not real uploadnig progress.
Then I tried to override transferTo method, print something (like transfered position), it's never got in and saw nothing in stdout.

Any useful advice? Thanks guys!

[slandelle]

There's actually many different ways of performing the body transfer, some even bypass memory and perform zero-copy and are not covered by the TransferCompletionHandler.

Known issue, low priority in my personal tasks. Typically something community could contribute.

[odiszapc]

@slandelle
Hi there, it looks like org.asynchttpclient.netty.request.body.BodyChunkedInput does not touch it's progress member at all. Are you ok with a one-line PR?:

@Override
public ByteBuf readChunk(ByteBufAllocator alloc) throws Exception {

    if (endOfInput)
        return null;

    ByteBuf buffer = alloc.buffer(chunkSize);
    Body.BodyState state = body.transferTo(buffer);
    progress += buffer.writerIndex();        // <<<<<< Increase progress
    switch (state) {
    case STOP:
        endOfInput = true;
        return buffer;
    case SUSPEND:
        // this will suspend the stream in ChunkedWriteHandler
        buffer.release();
        return null;
    case CONTINUE:
        return buffer;
    default:
        throw new IllegalStateException("Unknown state: " + state);
    }
}

After that correct values are passed to onByteSent. I think its normal that sometimes org.asynchttpclient.handler.TransferListener#onBytesSent receives amount = 0, we can just ignore such situations, the rest things looks good, current value grows monotonically:

onBytesSent, amount = 8192, current = 8192, total = 1000000
onBytesSent, amount = 8192, current = 16384, total = 1000000
onBytesSent, amount = 8192, current = 24576, total = 1000000
onBytesSent, amount = 73728, current = 98304, total = 1000000
onBytesSent, amount = 0, current = 98304, total = 1000000
onBytesSent, amount = 0, current = 98304, total = 1000000
onBytesSent, amount = 0, current = 98304, total = 1000000
onBytesSent, amount = 0, current = 98304, total = 1000000
onBytesSent, amount = 0, current = 98304, total = 1000000
onBytesSent, amount = 49152, current = 147456, total = 1000000
onBytesSent, amount = 0, current = 147456, total = 1000000
onBytesSent, amount = 0, current = 147456, total = 1000000
onBytesSent, amount = 0, current = 147456, total = 1000000
onBytesSent, amount = 0, current = 147456, total = 1000000
onBytesSent, amount = 0, current = 147456, total = 1000000
onBytesSent, amount = 49152, current = 196608, total = 1000000
onBytesSent, amount = 0, current = 196608, total = 1000000
onBytesSent, amount = 0, current = 196608, total = 1000000
onBytesSent, amount = 0, current = 196608, total = 1000000
onBytesSent, amount = 0, current = 196608, total = 1000000
onBytesSent, amount = 0, current = 196608, total = 1000000
onBytesSent, amount = 49152, current = 245760, total = 1000000
onBytesSent, amount = 0, current = 245760, total = 1000000
onBytesSent, amount = 0, current = 245760, total = 1000000

In my expirience the number of onByteSent call is always = total / 8192 (which is the default buffer size hardcoded in BodyChunkedInput static variable). You can see that a coupe of calls merged. Not famirlar with Netty, maybe merging happens due to non-blocking nature of network writes.

Any thoughts?

[odiszapc]

Fixed zeroes in org.asynchttpclient.netty.request.WriteProgressListener:

49        if (progress != lastLastProgress) {
50            progressAsyncHandler.onContentWriteProgress(progress - lastLastProgress, progress, total);
51        }

Result:

onBytesSent, amount = 8192, current = 8192, total = 1000000
onBytesSent, amount = 73728, current = 81920, total = 1000000
onBytesSent, amount = 49152, current = 131072, total = 1000000
onBytesSent, amount = 49152, current = 180224, total = 1000000
onBytesSent, amount = 49152, current = 229376, total = 1000000
onBytesSent, amount = 49152, current = 278528, total = 1000000
onBytesSent, amount = 49152, current = 327680, total = 1000000
onBytesSent, amount = 49152, current = 376832, total = 1000000
onBytesSent, amount = 49152, current = 425984, total = 1000000
onBytesSent, amount = 49152, current = 475136, total = 1000000
onBytesSent, amount = 49152, current = 524288, total = 1000000
onBytesSent, amount = 49152, current = 573440, total = 1000000
onBytesSent, amount = 49152, current = 622592, total = 1000000
onBytesSent, amount = 49152, current = 671744, total = 1000000
onBytesSent, amount = 49152, current = 720896, total = 1000000
onBytesSent, amount = 49152, current = 770048, total = 1000000
onBytesSent, amount = 49152, current = 819200, total = 1000000
onBytesSent, amount = 49152, current = 868352, total = 1000000
onBytesSent, amount = 49152, current = 917504, total = 1000000
onBytesSent, amount = 49152, current = 966656, total = 1000000
onBytesSent, amount = 33344, current = 1000000, total = 1000000

So, are you ok with that? Can I perform a PR?

[slandelle]

doh!

Yes, please go with a PR. Thanks a bunch!

[odiszapc]

Done: #1354

[odiszapc]

@slandelle excuse me, when can I expect this fix to be landed on Maven Central?