Add tarstream

distbuild/pkg/dist/state.go

@@ -0,0 +1,15 @@
+package dist
+
+import (
+	"gitlab.com/slon/shad-go/distbuild/pkg/build"
+	"gitlab.com/slon/shad-go/distbuild/pkg/proto"
+)
+
+type Cluster struct {
+	sourceFiles map[build.ID]map[proto.WorkerID]struct{}
+	artifacts   map[build.ID]map[proto.WorkerID]struct{}
+}
+
+func (c *Cluster) FindOptimalWorkers(task build.ID, sources, deps []build.ID) []proto.WorkerID {
+
+}

distbuild/pkg/proto/heartbeat.go

@@ -18,13 +18,15 @@ type JobResult struct {
 	Error *string
 }
 
+type WorkerID string
+
 type HeartbeatRequest struct {
 	// WorkerID задаёт персистентный идентификатор данного воркера.
 	//
 	// WorkerID так же выступает в качестве endpoint-а, к которому можно подключиться по HTTP.
 	//
 	// В наших тестов, идентификатор будет иметь вид "localhost:%d".
-	WorkerID string
+	WorkerID WorkerID
 
 	// ProcessID задаёт эфемерный идентификатор текущего процесса воркера.
 	//

distbuild/pkg/tarstream/stream.go

@@ -0,0 +1,96 @@
+package tarstream
+
+import (
+	"archive/tar"
+	"io"
+	"os"
+	"path/filepath"
+)
+
+func Send(dir string, w io.Writer) error {
+	tw := tar.NewWriter(w)
+
+	err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		rel, err := filepath.Rel(dir, path)
+		if err != nil {
+			return err
+		}
+
+		if rel == "." {
+			return nil
+		}
+
+		if info.IsDir() {
+			return tw.WriteHeader(&tar.Header{
+				Name:     rel,
+				Typeflag: tar.TypeDir,
+			})
+		} else {
+			h := &tar.Header{
+				Typeflag: tar.TypeReg,
+				Name:     rel,
+				Size:     info.Size(),
+				Mode:     int64(info.Mode()),
+			}
+
+			if err := tw.WriteHeader(h); err != nil {
+				return err
+			}
+
+			f, err := os.Open(path)
+			if err != nil {
+				return err
+			}
+			defer f.Close()
+
+			_, err = io.Copy(tw, f)
+			return err
+		}
+	})
+
+	if err != nil {
+		return err
+	}
+
+	return tw.Close()
+}
+
+func Receive(dir string, r io.Reader) error {
+	tr := tar.NewReader(r)
+
+	for {
+		h, err := tr.Next()
+		if err == io.EOF {
+			return nil
+		} else if err != nil {
+			return err
+		}
+
+		absPath := filepath.Join(dir, h.Name)
+
+		if h.Typeflag == tar.TypeDir {
+			if err := os.Mkdir(absPath, 0777); err != nil {
+				return err
+			}
+		} else {
+			writeFile := func() error {
+				f, err := os.OpenFile(absPath, os.O_CREATE|os.O_WRONLY, os.FileMode(h.Mode))
+				if err != nil {
+					return err
+				}
+				defer f.Close()
+
+				_, err = io.Copy(f, tr)
+				return err
+			}
+
+			if err := writeFile(); err != nil {
+				return err
+			}
+		}
+	}
+}

distbuild/pkg/tarstream/stream_test.go

@@ -0,0 +1,60 @@
+package tarstream
+
+import (
+	"bytes"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestTarStream(t *testing.T) {
+	tmpDir, err := ioutil.TempDir("", "tarstream")
+	require.NoError(t, err)
+
+	t.Logf("running inside %s", tmpDir)
+
+	from := filepath.Join(tmpDir, "from")
+	to := filepath.Join(tmpDir, "to")
+
+	require.NoError(t, os.Mkdir(from, 0777))
+	require.NoError(t, os.Mkdir(to, 0777))
+
+	var buf bytes.Buffer
+
+	require.NoError(t, os.Mkdir(filepath.Join(from, "a"), 0777))
+	require.NoError(t, os.MkdirAll(filepath.Join(from, "b", "c", "d"), 0777))
+	require.NoError(t, ioutil.WriteFile(filepath.Join(from, "a", "x.bin"), []byte("xxx"), 0777))
+	require.NoError(t, ioutil.WriteFile(filepath.Join(from, "b", "c", "y.txt"), []byte("yyy"), 0666))
+
+	require.NoError(t, Send(from, &buf))
+
+	require.NoError(t, Receive(to, &buf))
+
+	checkDir := func(path string) {
+		st, err := os.Stat(path)
+		require.NoError(t, err)
+		require.True(t, st.IsDir())
+	}
+
+	checkDir(filepath.Join(to, "a"))
+	checkDir(filepath.Join(to, "b", "c", "d"))
+
+	checkFile := func(path string, content []byte, mode os.FileMode) {
+		t.Helper()
+
+		st, err := os.Stat(path)
+		require.NoError(t, err)
+
+		require.Equal(t, mode.String(), st.Mode().String())
+
+		b, err := ioutil.ReadFile(path)
+		require.NoError(t, err)
+		require.Equal(t, content, b)
+	}
+
+	checkFile(filepath.Join(from, "a", "x.bin"), []byte("xxx"), 0755)
+	checkFile(filepath.Join(from, "b", "c", "y.txt"), []byte("yyy"), 0644)
+}