001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *   http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.eclipse.aether.spi.connector.transport;
020
021import java.io.IOException;
022import java.io.InputStream;
023import java.io.OutputStream;
024import java.nio.ByteBuffer;
025import java.util.concurrent.atomic.AtomicBoolean;
026
027import org.eclipse.aether.transfer.TransferCancelledException;
028
029import static java.util.Objects.requireNonNull;
030
031/**
032 * A skeleton implementation for custom transporters.
033 */
034public abstract class AbstractTransporter implements Transporter {
035
036    private final AtomicBoolean closed;
037
038    /**
039     * Enables subclassing.
040     */
041    protected AbstractTransporter() {
042        closed = new AtomicBoolean();
043    }
044
045    public void peek(PeekTask task) throws Exception {
046        requireNonNull(task, "task cannot be null");
047
048        failIfClosed(task);
049        implPeek(task);
050    }
051
052    /**
053     * Implements {@link #peek(PeekTask)}, gets only called if the transporter has not been closed.
054     *
055     * @param task The existence check to perform, must not be {@code null}.
056     * @throws Exception If the existence of the specified resource could not be confirmed.
057     */
058    protected abstract void implPeek(PeekTask task) throws Exception;
059
060    public void get(GetTask task) throws Exception {
061        requireNonNull(task, "task cannot be null");
062
063        failIfClosed(task);
064        implGet(task);
065    }
066
067    /**
068     * Implements {@link #get(GetTask)}, gets only called if the transporter has not been closed.
069     *
070     * @param task The download to perform, must not be {@code null}.
071     * @throws Exception If the transfer failed.
072     */
073    protected abstract void implGet(GetTask task) throws Exception;
074
075    /**
076     * Performs stream-based I/O for the specified download task and notifies the configured transport listener.
077     * Subclasses might want to invoke this utility method from within their {@link #implGet(GetTask)} to avoid
078     * boilerplate I/O code.
079     *
080     * @param task The download to perform, must not be {@code null}.
081     * @param is The input stream to download the data from, must not be {@code null}.
082     * @param close {@code true} if the supplied input stream should be automatically closed, {@code false} to leave the
083     *            stream open.
084     * @param length The size in bytes of the downloaded resource or {@code -1} if unknown, not to be confused with the
085     *            length of the supplied input stream which might be smaller if the download is resumed.
086     * @param resume {@code true} if the download resumes from {@link GetTask#getResumeOffset()}, {@code false} if the
087     *            download starts at the first byte of the resource.
088     * @throws IOException If the transfer encountered an I/O error.
089     * @throws TransferCancelledException If the transfer was cancelled.
090     */
091    protected void utilGet(GetTask task, InputStream is, boolean close, long length, boolean resume)
092            throws IOException, TransferCancelledException {
093        if (close) {
094            try (InputStream input = is;
095                    OutputStream os = task.newOutputStream(resume)) {
096                task.getListener().transportStarted(resume ? task.getResumeOffset() : 0L, length);
097                copy(os, input, task.getListener());
098            }
099        } else {
100            try (OutputStream os = task.newOutputStream(resume)) {
101                task.getListener().transportStarted(resume ? task.getResumeOffset() : 0L, length);
102                copy(os, is, task.getListener());
103            }
104        }
105    }
106
107    public void put(PutTask task) throws Exception {
108        requireNonNull(task, "task cannot be null");
109
110        failIfClosed(task);
111        implPut(task);
112    }
113
114    /**
115     * Implements {@link #put(PutTask)}, gets only called if the transporter has not been closed.
116     *
117     * @param task The upload to perform, must not be {@code null}.
118     * @throws Exception If the transfer failed.
119     */
120    protected abstract void implPut(PutTask task) throws Exception;
121
122    /**
123     * Performs stream-based I/O for the specified upload task and notifies the configured transport listener.
124     * Subclasses might want to invoke this utility method from within their {@link #implPut(PutTask)} to avoid
125     * boilerplate I/O code.
126     *
127     * @param task The upload to perform, must not be {@code null}.
128     * @param os The output stream to upload the data to, must not be {@code null}.
129     * @param close {@code true} if the supplied output stream should be automatically closed, {@code false} to leave
130     *            the stream open.
131     * @throws IOException If the transfer encountered an I/O error.
132     * @throws TransferCancelledException If the transfer was cancelled.
133     */
134    protected void utilPut(PutTask task, OutputStream os, boolean close)
135            throws IOException, TransferCancelledException {
136        if (close) {
137            try (OutputStream output = os;
138                    InputStream is = task.newInputStream()) {
139                task.getListener().transportStarted(0, task.getDataLength());
140                copy(output, is, task.getListener());
141            }
142        } else {
143            try (InputStream is = task.newInputStream()) {
144                task.getListener().transportStarted(0, task.getDataLength());
145                copy(os, is, task.getListener());
146            } finally {
147                os.flush();
148            }
149        }
150    }
151
152    public void close() {
153        if (closed.compareAndSet(false, true)) {
154            implClose();
155        }
156    }
157
158    /**
159     * Implements {@link #close()}, gets only called if the transporter has not already been closed.
160     */
161    protected abstract void implClose();
162
163    private void failIfClosed(TransportTask task) {
164        if (closed.get()) {
165            throw new IllegalStateException("transporter closed, cannot execute task " + task);
166        }
167    }
168
169    private static void copy(OutputStream os, InputStream is, TransportListener listener)
170            throws IOException, TransferCancelledException {
171        byte[] buffer = new byte[1024 * 32];
172        for (int read = is.read(buffer); read >= 0; read = is.read(buffer)) {
173            os.write(buffer, 0, read);
174            listener.transportProgressed(ByteBuffer.wrap(buffer, 0, read));
175        }
176    }
177}