streaming.go

Documentation: k8s.io/apimachinery/pkg/runtime/serializer/streaming

     1  /*
     2  Copyright 2015 The Kubernetes 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 streaming implements encoder and decoder for streams
    18  // of runtime.Objects over io.Writer/Readers.
    19  package streaming
    20  
    21  import (
    22  	"bytes"
    23  	"fmt"
    24  	"io"
    25  
    26  	"k8s.io/apimachinery/pkg/runtime"
    27  	"k8s.io/apimachinery/pkg/runtime/schema"
    28  )
    29  
    30  // Encoder is a runtime.Encoder on a stream.
    31  type Encoder interface {
    32  	// Encode will write the provided object to the stream or return an error. It obeys the same
    33  	// contract as runtime.VersionedEncoder.
    34  	Encode(obj runtime.Object) error
    35  }
    36  
    37  // Decoder is a runtime.Decoder from a stream.
    38  type Decoder interface {
    39  	// Decode will return io.EOF when no more objects are available.
    40  	Decode(defaults *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error)
    41  	// Close closes the underlying stream.
    42  	Close() error
    43  }
    44  
    45  // Serializer is a factory for creating encoders and decoders that work over streams.
    46  type Serializer interface {
    47  	NewEncoder(w io.Writer) Encoder
    48  	NewDecoder(r io.ReadCloser) Decoder
    49  }
    50  
    51  type decoder struct {
    52  	reader    io.ReadCloser
    53  	decoder   runtime.Decoder
    54  	buf       []byte
    55  	maxBytes  int
    56  	resetRead bool
    57  }
    58  
    59  // NewDecoder creates a streaming decoder that reads object chunks from r and decodes them with d.
    60  // The reader is expected to return ErrShortRead if the provided buffer is not large enough to read
    61  // an entire object.
    62  func NewDecoder(r io.ReadCloser, d runtime.Decoder) Decoder {
    63  	return &decoder{
    64  		reader:   r,
    65  		decoder:  d,
    66  		buf:      make([]byte, 1024),
    67  		maxBytes: 16 * 1024 * 1024,
    68  	}
    69  }
    70  
    71  var ErrObjectTooLarge = fmt.Errorf("object to decode was longer than maximum allowed size")
    72  
    73  // Decode reads the next object from the stream and decodes it.
    74  func (d *decoder) Decode(defaults *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
    75  	base := 0
    76  	for {
    77  		n, err := d.reader.Read(d.buf[base:])
    78  		if err == io.ErrShortBuffer {
    79  			if n == 0 {
    80  				return nil, nil, fmt.Errorf("got short buffer with n=0, base=%d, cap=%d", base, cap(d.buf))
    81  			}
    82  			if d.resetRead {
    83  				continue
    84  			}
    85  			// double the buffer size up to maxBytes
    86  			if len(d.buf) < d.maxBytes {
    87  				base += n
    88  				d.buf = append(d.buf, make([]byte, len(d.buf))...)
    89  				continue
    90  			}
    91  			// must read the rest of the frame (until we stop getting ErrShortBuffer)
    92  			d.resetRead = true
    93  			return nil, nil, ErrObjectTooLarge
    94  		}
    95  		if err != nil {
    96  			return nil, nil, err
    97  		}
    98  		if d.resetRead {
    99  			// now that we have drained the large read, continue
   100  			d.resetRead = false
   101  			continue
   102  		}
   103  		base += n
   104  		break
   105  	}
   106  	return d.decoder.Decode(d.buf[:base], defaults, into)
   107  }
   108  
   109  func (d *decoder) Close() error {
   110  	return d.reader.Close()
   111  }
   112  
   113  type encoder struct {
   114  	writer  io.Writer
   115  	encoder runtime.Encoder
   116  	buf     *bytes.Buffer
   117  }
   118  
   119  // NewEncoder returns a new streaming encoder.
   120  func NewEncoder(w io.Writer, e runtime.Encoder) Encoder {
   121  	return &encoder{
   122  		writer:  w,
   123  		encoder: e,
   124  		buf:     &bytes.Buffer{},
   125  	}
   126  }
   127  
   128  // Encode writes the provided object to the nested writer.
   129  func (e *encoder) Encode(obj runtime.Object) error {
   130  	if err := e.encoder.Encode(obj, e.buf); err != nil {
   131  		return err
   132  	}
   133  	_, err := e.writer.Write(e.buf.Bytes())
   134  	e.buf.Reset()
   135  	return err
   136  }
   137
View as plain text