2 Copyright 2015 The Kubernetes Authors.
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
8 http://www.apache.org/licenses/LICENSE-2.0
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.
17 // Package streaming implements encoder and decoder for streams
18 // of runtime.Objects over io.Writer/Readers.
26 "k8s.io/apimachinery/pkg/runtime"
27 "k8s.io/apimachinery/pkg/runtime/schema"
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
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.
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
53 decoder runtime.Decoder
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
62 func NewDecoder(r io.ReadCloser, d runtime.Decoder) Decoder {
66 buf: make([]byte, 1024),
67 maxBytes: 16 * 1024 * 1024,
71 var ErrObjectTooLarge = fmt.Errorf("object to decode was longer than maximum allowed size")
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) {
77 n, err := d.reader.Read(d.buf[base:])
78 if err == io.ErrShortBuffer {
80 return nil, nil, fmt.Errorf("got short buffer with n=0, base=%d, cap=%d", base, cap(d.buf))
85 // double the buffer size up to maxBytes
86 if len(d.buf) < d.maxBytes {
88 d.buf = append(d.buf, make([]byte, len(d.buf))...)
91 // must read the rest of the frame (until we stop getting ErrShortBuffer)
94 return nil, nil, ErrObjectTooLarge
100 // now that we have drained the large read, continue
107 return d.decoder.Decode(d.buf[:base], defaults, into)
110 func (d *decoder) Close() error {
111 return d.reader.Close()
114 type encoder struct {
116 encoder runtime.Encoder
120 // NewEncoder returns a new streaming encoder.
121 func NewEncoder(w io.Writer, e runtime.Encoder) Encoder {
125 buf: &bytes.Buffer{},
129 // Encode writes the provided object to the nested writer.
130 func (e *encoder) Encode(obj runtime.Object) error {
131 if err := e.encoder.Encode(obj, e.buf); err != nil {
134 _, err := e.writer.Write(e.buf.Bytes())