001/**
002 * Java Web Archive Toolkit - Software to read and validate ARC, WARC
003 * and GZip files. (http://jwat.org/)
004 * Copyright 2011-2012 Netarkivet.dk (http://netarkivet.dk/)
005 *
006 * Licensed under the Apache License, Version 2.0 (the "License");
007 * you may not use this file except in compliance with the License.
008 * 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, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.jwat.common;
019
020import java.io.BufferedInputStream;
021import java.io.Closeable;
022import java.io.IOException;
023import java.io.InputStream;
024import java.security.DigestInputStream;
025import java.security.MessageDigest;
026import java.security.NoSuchAlgorithmException;
027
028/**
029 * This class makes the archived payload of an ARC/WARC record accessible
030 * through a stream. The stream is fixed length so only the payload is
031 * available.
032 *
033 * @author lbihanic, selghissassi, nicl
034 */
035public class Payload implements Closeable {
036
037    /** Payload content. */
038    private static final int BUFFER_SIZE = 8192;
039
040    /** Has record been closed before. */
041    protected boolean bClosed;
042
043    /** Payload length. */
044    protected long length;
045
046    /** Base stream used to limit payload access to only the payload and
047     * not any record data beyond that. Also detects unexpected EOF. */
048    protected FixedLengthInputStream in_fl;
049
050    /** Message digest object. */
051    protected MessageDigest md;
052
053    /** Digest bytes. */
054    protected byte[] digest;
055
056    /** Automatic digesting of payload input stream. */
057    protected DigestInputStream in_digest;
058
059    /** Boolean indicating no such algorithm exception under initialization. */
060    protected boolean bNoSuchAlgorithmException;
061
062    /** Payload content. */
063    protected BufferedInputStream in_buffered;
064
065    /** Pushback input stream exposed to the outside, usable by payload processors. */
066    protected ByteCountingPushBackInputStream in_pb_exposed;
067
068    /** Pushback size. */
069    protected int pushback_size;
070
071    /** Header wrapped payload, if present. */
072    protected PayloadWithHeaderAbstract payloadHeaderWrapped;
073
074    /** Handler called when this payloads stream has been fully consumed. */
075    protected PayloadOnClosedHandler onClosedHandler;
076
077    /**
078     * Non public constructor.
079     */
080    protected Payload() {
081    }
082
083    /**
084     * Creates new <code>ArcPayload</code> instance.
085     * @param in the input stream to parse.
086     * @param length payload length.
087     * @param pushback_size   pushback size
088     * @param digestAlgorithm digest algorithm to use on payload or null
089     * @return the payload
090     * @throws IOException if an I/O error occurs while initializing
091     */
092    public static Payload processPayload(InputStream in, long length,
093            int pushback_size, String digestAlgorithm) throws IOException {
094        if (in == null) {
095            throw new IllegalArgumentException(
096                    "The inputstream 'in' is null");
097        }
098        if (length < 0) {
099            throw new IllegalArgumentException(
100                    "The 'length' is less than zero: " + length);
101        }
102        if (pushback_size <= 0) {
103            throw new IllegalArgumentException(
104                    "The 'pushback_size' is less than or equal to zero: " +
105                            pushback_size);
106        }
107        Payload pl = new Payload();
108        pl.length = length;
109        pl.pushback_size = pushback_size;
110        pl.in_fl = new FixedLengthInputStream(in, length);
111        /*
112         * Block Digest.
113         */
114        if (digestAlgorithm != null) {
115            try {
116                pl.md = MessageDigest.getInstance(digestAlgorithm);
117            } catch (NoSuchAlgorithmException e) {
118                pl.bNoSuchAlgorithmException = true;
119            }
120        }
121        if (pl.md != null) {
122            pl.in_digest = new DigestInputStreamNoSkip(pl.in_fl, pl.md);
123            pl.in_buffered = new BufferedInputStream(pl.in_digest, BUFFER_SIZE);
124        } else {
125            pl.in_buffered = new BufferedInputStream(pl.in_fl, BUFFER_SIZE);
126        }
127        /*
128         * Ensure close() is not called on the payload stream!
129         */
130        pl.in_pb_exposed = new ByteCountingPushBackInputStream(pl.in_buffered, pushback_size) {
131            @Override
132            public void close() throws IOException {
133            }
134        };
135        return pl;
136    }
137
138    /**
139     * Set optional handler to be called when payload is closed.
140     * This method should not be called by the payload consumer.
141     * @param onClosedHandler on closed handler implementation
142     */
143    public void setOnClosedHandler(PayloadOnClosedHandler onClosedHandler) {
144        this.onClosedHandler = onClosedHandler;
145    }
146
147    /**
148     * Returns the calculated digest.
149     * @return the calculated digest
150     */
151    public byte[] getDigest() {
152        if (digest == null && md != null) {
153            digest = md.digest();
154        }
155        return digest;
156    }
157
158    /**
159     * Get payload total length.
160     * @return payload total length
161     */
162    public long getTotalLength() {
163        return length;
164    }
165
166    /**
167     * Get the number of unavailable bytes missing due to unexpected EOF.
168     * This method always returns <code>0</code> as long as the stream is open.
169     * @return number of unavailable bytes missing due to unexpected EOF
170     * @throws IOException if an I/O error occurs calling available method on stream
171     */
172    public long getUnavailable() throws IOException {
173        return in_fl.available();
174    }
175
176    /**
177     * Get pushback buffer size.
178     * @return pushback buffer size
179     */
180    public int getPushbackSize() {
181        return pushback_size;
182    }
183
184    /**
185     * Set <code>PayloadHeaderWrapper</code> object in case of recognized payload content.
186     * This method should not be called by the payload consumer.
187     * @param payloadHeaderWrapped <code>PayloadHeaderWrapper</code> object
188     */
189    public void setPayloadHeaderWrapped(PayloadWithHeaderAbstract payloadHeaderWrapped) {
190        this.payloadHeaderWrapped = payloadHeaderWrapped;
191    }
192
193    /**
194     * Get the <code>PayloadHeaderWrapper</code> object associated with this payload.
195     * @return <code>PayloadHeaderWrapper</code> object or null
196     */
197    public PayloadWithHeaderAbstract getPayloadHeaderWrapped() {
198        return payloadHeaderWrapped;
199    }
200
201    /**
202     * Get <code>InputStream</code> to read the complete payload even though
203     * a http response header may have been read.
204     * @return <code>InputStream</code> to read payload data (in)directly.
205     */
206    public InputStream getInputStreamComplete() {
207        if (payloadHeaderWrapped != null) {
208            return payloadHeaderWrapped.getInputStreamComplete();
209        } else {
210            return in_pb_exposed;
211        }
212    }
213
214    /**
215     * Get <code>InputStream</code> to read the payload directly from the
216     * source bypassing any existing <code>HttpResponse</code> object.
217     * A parsed http response header will not be accessible through this
218     * stream.
219     * @return <code>InputStream</code> to read payload data.
220     */
221    public ByteCountingPushBackInputStream getInputStream() {
222        if (payloadHeaderWrapped != null) {
223            return payloadHeaderWrapped.getPayloadInputStream();
224        } else {
225            return in_pb_exposed;
226        }
227    }
228
229    /**
230     * Get payload remaining length.
231     * @return payload remaining length
232     * @throws IOException if an I/O error occurs calling available method on stream
233     */
234    public long getRemaining() throws IOException {
235        if (payloadHeaderWrapped != null) {
236            return payloadHeaderWrapped.getPayloadInputStream().available();
237        } else {
238            return in_pb_exposed.available();
239        }
240    }
241
242    /**
243     * Check to see if the payload has been closed.
244     * @return boolean indicating whether this payload is closed or not
245     */
246    public boolean isClosed() {
247        return bClosed;
248    }
249
250    /**
251     * Closes the this payload stream, skipping unread bytes in the process.
252     * @throws IOException if an I/O error occurs in the closing process
253     */
254    public void close() throws IOException {
255        if (!bClosed) {
256            if (payloadHeaderWrapped != null) {
257                payloadHeaderWrapped.close();
258            }
259            if (md != null) {
260                // Skip remaining unread bytes to ensure payload is completely
261                // digested. Skipping because the DigestInputStreamNoSkip
262                // has been altered to read when skipping.
263                while (in_digest.skip(length) > 0) {
264                }
265            }
266            if (in_buffered != null) {
267                in_buffered.close();
268                in_buffered = null;
269            }
270            if (onClosedHandler != null) {
271                onClosedHandler.payloadClosed();
272                onClosedHandler = null;
273            }
274            bClosed = true;
275        }
276    }
277
278}