Function newUpstream

pkg/devspace/sync/upstream.go:71–120

newUpstream creates a new upstream handler with the given parameters

(reader io.ReadCloser, writer io.WriteCloser, sync *Sync)

// newUpstream creates a new upstream handler with the given parameters
func newUpstream(reader io.ReadCloser, writer io.WriteCloser, sync *Sync) (*upstream, error) {
	var (
		clientReader io.Reader = reader
		clientWriter io.Writer = writer
	)

	// Apply limits if specified
	if sync.Options.DownstreamLimit > 0 {
		limitedReader := shapeio.NewReader(reader)
		limitedReader.SetRateLimit(float64(sync.Options.DownstreamLimit))
		clientReader = limitedReader
	}
	if sync.Options.UpstreamLimit > 0 {
		limitedWriter := shapeio.NewWriter(writer)
		limitedWriter.SetRateLimit(float64(sync.Options.UpstreamLimit))
		clientWriter = limitedWriter
	}

	// Create client
	conn, err := util.NewClientConnection(clientReader, clientWriter)
	if err != nil {
		return nil, errors.Wrap(err, "new client connection")
	}

	// Create combined exclude paths
	excludePaths := make([]string, 0, len(sync.Options.ExcludePaths)+len(sync.Options.UploadExcludePaths))
	excludePaths = append(excludePaths, sync.Options.ExcludePaths...)
	excludePaths = append(excludePaths, sync.Options.UploadExcludePaths...)

	ignoreMatcher, err := ignoreparser.CompilePaths(excludePaths, sync.log)
	if err != nil {
		return nil, errors.Wrap(err, "compile paths")
	}

	return &upstream{
		events:      make(chan notify.EventInfo, 1000), // High buffer size so we don't miss any fsevents if there are a lot of changes
		eventBuffer: make([]notify.EventInfo, 0, 64),
		symlinks:    make(map[string]*Symlink),
		sync:        sync,
		isBusy:      true,

		reader: reader,
		writer: writer,
		client: remote.NewUpstreamClient(conn),

		ignoreMatcher: ignoreMatcher,

		conn: conn,
	}, nil
}

func (u *upstream) IsBusy() bool {
	u.isBusyMutex.Lock()