OKHTTP 3跟踪多部分上载进度

Question

private static final String IMGUR_CLIENT_ID = "...";
private static final MediaType MEDIA_TYPE_PNG = MediaType.parse("image/png");

private final OkHttpClient client = new OkHttpClient();

public void run() throws Exception {
    // Use the imgur image upload API as documented at https://api.imgur.com/endpoints/image
    RequestBody requestBody = new MultipartBody.Builder()
            .setType(MultipartBody.FORM)
            .addFormDataPart("title", "Square Logo")
            .addFormDataPart("image", "logo-square.png",
                    RequestBody.create(MEDIA_TYPE_PNG, new File("website/static/logo-square.png")))
            .build();

    Request request = new Request.Builder()
            .header("Authorization", "Client-ID " + IMGUR_CLIENT_ID)
            .url("https://api.imgur.com/3/image")
            .post(requestBody)
            .build();

    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) throw new IOException("Unexpected code " + response);

    System.out.println(response.body().string());
}

Answer 1

我无法得到任何对我有效的答案。问题是，在图像上传之前，进度将运行到100%，暗示在数据通过网络发送之前，某个缓冲区已经被填满。经过一些研究，我发现确实是这样，这个缓冲区就是套接字发送缓冲区。向OkHttpClient提供一个SocketFactory终于奏效了。我的科特林密码如下...

首先，和其他人一样，我有一个CountingRequestBody，用于包装MultipartBody。

class CountingRequestBody(var delegate: RequestBody, private var listener: (max: Long, value: Long) -> Unit): RequestBody() {

    override fun contentType(): MediaType? {
        return delegate.contentType()
    }

    override fun contentLength(): Long {
        try {
            return delegate.contentLength()
        } catch (e: IOException) {
            e.printStackTrace()
        }
        return -1
    }

    override fun writeTo(sink: BufferedSink) {
        val countingSink = CountingSink(sink)
        val bufferedSink = Okio.buffer(countingSink)
        delegate.writeTo(bufferedSink)
        bufferedSink.flush()
    }

    inner class CountingSink(delegate: Sink): ForwardingSink(delegate) {
        private var bytesWritten: Long = 0

        override fun write(source: Buffer, byteCount: Long) {
            super.write(source, byteCount)
            bytesWritten += byteCount
            listener(contentLength(), bytesWritten)
        }
    }
}

我在Retrofit2中使用这个。一般用法如下：

val builder = MultipartBody.Builder()
// Add stuff to the MultipartBody via the Builder

val body = CountingRequestBody(builder.build()) { max, value ->
      // Progress your progress, or send it somewhere else.
}

class ProgressFriendlySocketFactory(private val sendBufferSize: Int = DEFAULT_BUFFER_SIZE) : SocketFactory() {

    override fun createSocket(): Socket {
        return setSendBufferSize(Socket())
    }

    override fun createSocket(host: String, port: Int): Socket {
        return setSendBufferSize(Socket(host, port))
    }

    override fun createSocket(host: String, port: Int, localHost: InetAddress, localPort: Int): Socket {
        return setSendBufferSize(Socket(host, port, localHost, localPort))
    }

    override fun createSocket(host: InetAddress, port: Int): Socket {
        return setSendBufferSize(Socket(host, port))
    }

    override fun createSocket(address: InetAddress, port: Int, localAddress: InetAddress, localPort: Int): Socket {
        return setSendBufferSize(Socket(address, port, localAddress, localPort))
    }

    private fun setSendBufferSize(socket: Socket): Socket {
        socket.sendBufferSize = sendBufferSize
        return socket
    }

    companion object {
        const val DEFAULT_BUFFER_SIZE = 2048
    }
}

val clientBuilder = OkHttpClient.Builder()
    .socketFactory(ProgressFriendlySocketFactory())

正如其他人所提到的，记录请求正文可能会影响这一点，并导致数据被多次读取。要么不记录正文，要么为CountingRequestBody关闭它。为此，我编写了自己的HttpLoggingInterceptor，它解决了这个问题和其他问题（比如记录MultipartBody）。但这超出了这个问题的范围。

if(requestBody is CountingRequestBody) {
  // don't log the body in production
}

其他问题与MockWebServer有关。我有一个使用MockWebServer和json文件的口味，这样我的应用程序就可以在没有网络的情况下运行，这样我就可以在没有负担的情况下进行测试。为了让这段代码工作，调度程序需要读取正文数据。我创建这个调度程序就是为了做到这一点。然后它将分派转发给另一个分派器，例如默认的QueueDispatcher。

class BodyReadingDispatcher(val child: Dispatcher): Dispatcher() {

    override fun dispatch(request: RecordedRequest?): MockResponse {
        val body = request?.body
        if(body != null) {
            val sink = ByteArray(1024)
            while(body.read(sink) >= 0) {
                Thread.sleep(50) // change this time to work for you
            }
        }
        val response = child.dispatch(request)
        return response
    }
}

您可以在MockWebServer中将其用作：

var server = MockWebServer()
server.setDispatcher(BodyReadingDispatcher(QueueDispatcher()))