transceivr/src/main/java/com/acdp/transceivr/LoadTools.java

package com.acdp.transceivr;

import okhttp3.*;
import okio.BufferedSink;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;

public class LoadTools {
    public static final MediaType MEDIA_TYPE_BINARY
            = MediaType.parse("application/octet-stream");
    public static final MediaType MEDIA_TYPE_JSON
            = MediaType.get("application/json; charset=utf-8");
    private final static org.slf4j.Logger logger = LoggerFactory.getLogger(LoadTools.class);

    private static void finishTransfer(Transfer t) {
        t.finishedAT=System.currentTimeMillis();
        t.finished=true;
    }

    private static void download(Transfer t, ProgressListener progressListener) throws IOException {
        Request request = new Request.Builder()
            .url(t.from)
            .build();

        OkHttpClient client = new OkHttpClient.Builder()
            .addNetworkInterceptor(chain -> {
                Response originalResponse = chain.proceed(chain.request());
                return originalResponse.newBuilder()
                        .body(new ProgressResponseBody(originalResponse.body(), progressListener))
                        .build();
            })
            .build();
        var call = client.newCall(request);
        t.call = call;
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                finishTransfer(t);
                t.uploadError = e.toString();
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                t.downloadCode = response.code();
                if (!response.isSuccessful()) {
                    finishTransfer(t);
                    return;
                }
                try {
                    upload(t, response.body().byteStream());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    private static void upload(Transfer t, InputStream is) throws IOException {
        RequestBody requestBody = new RequestBody() {
            @Override
            public MediaType contentType() {
                return MEDIA_TYPE_BINARY;
            }

            @Override
            public void writeTo(BufferedSink sink) throws IOException {
                is.transferTo(sink.outputStream());
            }
        };

        OkHttpClient client = new OkHttpClient();
        var requestBuilder = new Request.Builder().url(t.to);
        if(t.toMethod.equals("PUT"))
            requestBuilder.put(requestBody);
        else
            requestBuilder.post(requestBody);

        var request=requestBuilder.build();
        client.newCall(request).enqueue(
            new Callback() {
                @Override
                public void onFailure(Call call, IOException e) {
                    t.uploadError = e.toString();
                    finishTransfer(t);
                }

                @Override
                public void onResponse(Call call, Response response) throws IOException {
                    if (!response.isSuccessful()) {
                        t.uploadError = response.toString();
                        finishTransfer(t);
                    }
                    t.uploadCode = response.code();
                    t.uploadResponseBody = response.body().string();
                    t.uploadDone = true;
                    if (t.cpeeCallback != null) {
                        try {
                            OkHttpClient client = new OkHttpClient();
                            Request request = new Request.Builder()
                                    .url(t.cpeeCallback + "/")
                                    .put(RequestBody.create(MEDIA_TYPE_JSON, "[{}]"))
                                    .build();
                            logger.info("putting to {}", t.cpeeCallback);
                            Response res = client.newCall(request).execute();
                            logger.info("result {}", res.toString());
                            t.cpeeCallbackResult = res.body().string();
                            t.cpeeCallbackCode = res.code();
                            t.cpeeCallbackMessage = res.message();
                        } catch (Exception e) {
                            logger.warn("cpee callback failed", e);
                            t.cpeeCallbackError = e.getMessage();
                            finishTransfer(t);
                        }
                    }
                    else {
                        finishTransfer(t);
                    }
                }
            }
        );
    }

    public static void startTransfer(Transfer t) throws IOException {
        final ProgressListener progressListener = new ProgressListener() {
            boolean firstUpdate = true;
            @Override
            public void update(long bytesRead, long contentLength, boolean done) {
                if (done) {
                    t.downloadDone = true;
                } else {
                    if (firstUpdate) {
                        firstUpdate = false;
                        if (contentLength == -1) {
                            t.contentLength = -1;
                        } else {
                            t.contentLength = contentLength;
                        }
                    }
                    t.bytesRead = bytesRead;
                    t.currentTS = System.currentTimeMillis();
                }
            }
        };
        download(t, progressListener);
    }
}