1	/*
     2	Copyright 2014 The Perkeep AUTHORS
     3	
     4	Licensed under the Apache License, Version 2.0 (the "License");
     5	you may not use this file except in compliance with the License.
     6	You may obtain a copy of the License at
     7	
     8	     http://www.apache.org/licenses/LICENSE-2.0
     9	
    10	Unless required by applicable law or agreed to in writing, software
    11	distributed under the License is distributed on an "AS IS" BASIS,
    12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13	See the License for the specific language governing permissions and
    14	limitations under the License.
    15	*/
    16	
    17	package blobpacked
    18	
    19	import (
    20		"context"
    21		"io"
    22	
    23		"perkeep.org/pkg/blob"
    24	)
    25	
    26	// Ensure we implement the optional interface correctly.
    27	var _ blob.SubFetcher = (*storage)(nil)
    28	
    29	// SubFetch returns part of a blob.
    30	// The caller must close the returned io.ReadCloser.
    31	// The Reader may return fewer than 'length' bytes. Callers should
    32	// check. The returned error should be os.ErrNotExist if the blob
    33	// doesn't exist.
    34	func (s *storage) SubFetch(ctx context.Context, ref blob.Ref, offset, length int64) (io.ReadCloser, error) {
    35		// TODO: pass ctx to more calls within this method.
    36		m, err := s.getMetaRow(ref)
    37		if err != nil {
    38			return nil, err
    39		}
    40		if m.isPacked() {
    41			length, err = capOffsetLength(m.size, offset, length)
    42			if err != nil {
    43				return nil, err
    44			}
    45			// get the blob from the large subfetcher
    46			return s.large.SubFetch(ctx, m.largeRef, int64(m.largeOff)+offset, length)
    47		}
    48		if sf, ok := s.small.(blob.SubFetcher); ok {
    49			rc, err := sf.SubFetch(ctx, ref, offset, length)
    50			if err != blob.ErrUnimplemented {
    51				return rc, err
    52			}
    53		}
    54		rc, size, err := s.small.Fetch(ctx, ref)
    55		if err != nil {
    56			return rc, err
    57		}
    58		length, err = capOffsetLength(size, offset, length)
    59		if err != nil {
    60			rc.Close()
    61			return nil, err
    62		}
    63		if offset != 0 {
    64			if _, err = io.CopyN(io.Discard, rc, offset); err != nil {
    65				rc.Close()
    66				return nil, err
    67			}
    68		}
    69		return struct {
    70			io.Reader
    71			io.Closer
    72		}{io.LimitReader(rc, length), rc}, nil
    73	}
    74	
    75	func capOffsetLength(size uint32, offset, length int64) (newLength int64, err error) {
    76		if offset < 0 || length < 0 {
    77			return 0, blob.ErrNegativeSubFetch
    78		}
    79		if offset > int64(size) {
    80			return 0, blob.ErrOutOfRangeOffsetSubFetch
    81		}
    82		if over := (offset + length) - int64(size); over > 0 {
    83			length -= over
    84		}
    85		return length, nil
    86	}