* Inital routes to git refs api * Git refs API implementation * Update swagger * Fix copyright * Make swagger happy add basic test * Fix test * Fix test again :)
148 lines
3.2 KiB
Go
148 lines
3.2 KiB
Go
package sideband
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
|
|
"gopkg.in/src-d/go-git.v4/plumbing/format/pktline"
|
|
)
|
|
|
|
// ErrMaxPackedExceeded returned by Read, if the maximum packed size is exceeded
|
|
var ErrMaxPackedExceeded = errors.New("max. packed size exceeded")
|
|
|
|
// Progress where the progress information is stored
|
|
type Progress interface {
|
|
io.Writer
|
|
}
|
|
|
|
// Demuxer demultiplexes the progress reports and error info interleaved with the
|
|
// packfile itself.
|
|
//
|
|
// A sideband has three different channels the main one, called PackData, contains
|
|
// the packfile data; the ErrorMessage channel, that contains server errors; and
|
|
// the last one, ProgressMessage channel, containing information about the ongoing
|
|
// task happening in the server (optional, can be suppressed sending NoProgress
|
|
// or Quiet capabilities to the server)
|
|
//
|
|
// In order to demultiplex the data stream, method `Read` should be called to
|
|
// retrieve the PackData channel, the incoming data from the ProgressMessage is
|
|
// written at `Progress` (if any), if any message is retrieved from the
|
|
// ErrorMessage channel an error is returned and we can assume that the
|
|
// connection has been closed.
|
|
type Demuxer struct {
|
|
t Type
|
|
r io.Reader
|
|
s *pktline.Scanner
|
|
|
|
max int
|
|
pending []byte
|
|
|
|
// Progress is where the progress messages are stored
|
|
Progress Progress
|
|
}
|
|
|
|
// NewDemuxer returns a new Demuxer for the given t and read from r
|
|
func NewDemuxer(t Type, r io.Reader) *Demuxer {
|
|
max := MaxPackedSize64k
|
|
if t == Sideband {
|
|
max = MaxPackedSize
|
|
}
|
|
|
|
return &Demuxer{
|
|
t: t,
|
|
r: r,
|
|
max: max,
|
|
s: pktline.NewScanner(r),
|
|
}
|
|
}
|
|
|
|
// Read reads up to len(p) bytes from the PackData channel into p, an error can
|
|
// be return if an error happens when reading or if a message is sent in the
|
|
// ErrorMessage channel.
|
|
//
|
|
// When a ProgressMessage is read, is not copy to b, instead of this is written
|
|
// to the Progress
|
|
func (d *Demuxer) Read(b []byte) (n int, err error) {
|
|
var read, req int
|
|
|
|
req = len(b)
|
|
for read < req {
|
|
n, err := d.doRead(b[read:req])
|
|
read += n
|
|
|
|
if err != nil {
|
|
return read, err
|
|
}
|
|
}
|
|
|
|
return read, nil
|
|
}
|
|
|
|
func (d *Demuxer) doRead(b []byte) (int, error) {
|
|
read, err := d.nextPackData()
|
|
size := len(read)
|
|
wanted := len(b)
|
|
|
|
if size > wanted {
|
|
d.pending = read[wanted:]
|
|
}
|
|
|
|
if wanted > size {
|
|
wanted = size
|
|
}
|
|
|
|
size = copy(b, read[:wanted])
|
|
return size, err
|
|
}
|
|
|
|
func (d *Demuxer) nextPackData() ([]byte, error) {
|
|
content := d.getPending()
|
|
if len(content) != 0 {
|
|
return content, nil
|
|
}
|
|
|
|
if !d.s.Scan() {
|
|
if err := d.s.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return nil, io.EOF
|
|
}
|
|
|
|
content = d.s.Bytes()
|
|
|
|
size := len(content)
|
|
if size == 0 {
|
|
return nil, nil
|
|
} else if size > d.max {
|
|
return nil, ErrMaxPackedExceeded
|
|
}
|
|
|
|
switch Channel(content[0]) {
|
|
case PackData:
|
|
return content[1:], nil
|
|
case ProgressMessage:
|
|
if d.Progress != nil {
|
|
_, err := d.Progress.Write(content[1:])
|
|
return nil, err
|
|
}
|
|
case ErrorMessage:
|
|
return nil, fmt.Errorf("unexpected error: %s", content[1:])
|
|
default:
|
|
return nil, fmt.Errorf("unknown channel %s", content)
|
|
}
|
|
|
|
return nil, nil
|
|
}
|
|
|
|
func (d *Demuxer) getPending() (b []byte) {
|
|
if len(d.pending) == 0 {
|
|
return nil
|
|
}
|
|
|
|
content := d.pending
|
|
d.pending = nil
|
|
|
|
return content
|
|
}
|