Implement OperationService::UserMergeBranch

2 files changed, 69 insertions, 28 deletions

diff --git a/internal/rubyserver/proxy.go b/internal/rubyserver/proxy.go
new file mode 100644
index 000000000..ee6ee137d
--- /dev/null
+++ b/internal/rubyserver/proxy.go
@@ -0,0 +1,69 @@
+package rubyserver
+
+import (
+	"context"
+	"io"
+
+	pb "gitlab.com/gitlab-org/gitaly-proto/go"
+	"gitlab.com/gitlab-org/gitaly/internal/helper"
+
+	"google.golang.org/grpc/metadata"
+)
+
+// SetHeaders adds headers that tell gitaly-ruby the full path to the repository.
+func SetHeaders(ctx context.Context, repo *pb.Repository) (context.Context, error) {
+	repoPath, err := helper.GetPath(repo)
+	if err != nil {
+		return nil, err
+	}
+
+	md := metadata.Pairs(repoPathHeader, repoPath, glRepositoryHeader, repo.GlRepository)
+	newCtx := metadata.NewOutgoingContext(ctx, md)
+	return newCtx, nil
+}
+
+// Proxy calls recvSend until it receives an error. The error is returned
+// to the caller unless it is io.EOF.
+func Proxy(recvSend func() error) (err error) {
+	for err == nil {
+		err = recvSend()
+	}
+
+	if err == io.EOF {
+		err = nil
+	}
+	return err
+}
+
+// CloseSender captures the CloseSend method from gRPC streams.
+type CloseSender interface {
+	CloseSend() error
+}
+
+// ProxyBidi works like Proxy but runs multiple callbacks simultaneously.
+// It returns immediately if proxying one of the callbacks fails. If the
+// response stream is done, ProxyBidi returns immediately without waiting
+// for the client stream to finish proxying.
+func ProxyBidi(requestFunc func() error, requestStream CloseSender, responseFunc func() error) error {
+	requestErr := make(chan error, 1)
+	go func() {
+		requestErr <- Proxy(requestFunc)
+	}()
+
+	responseErr := make(chan error, 1)
+	go func() {
+		responseErr <- Proxy(responseFunc)
+	}()
+
+	for {
+		select {
+		case err := <-requestErr:
+			if err != nil {
+				return err
+			}
+			requestStream.CloseSend()
+		case err := <-responseErr:
+			return err
+		}
+	}
+}
diff --git a/internal/rubyserver/rubyserver.go b/internal/rubyserver/rubyserver.go
index 434add1c3..d0a77b488 100644
--- a/internal/rubyserver/rubyserver.go
+++ b/internal/rubyserver/rubyserver.go
@@ -2,7 +2,6 @@ package rubyserver
 
 import (
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net"
 	"os"
@@ -14,7 +13,6 @@ import (
 
 	"gitlab.com/gitlab-org/gitaly/internal/command"
 	"gitlab.com/gitlab-org/gitaly/internal/config"
-	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/internal/supervisor"
 	"gitlab.com/gitlab-org/gitaly/streamio"
 
@@ -23,7 +21,6 @@ import (
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/metadata"
 )
 
 const (
@@ -192,28 +189,3 @@ func dialOptions() []grpc.DialOption {
 		}),
 	}
 }
-
-// SetHeaders adds headers that tell gitaly-ruby the full path to the repository.
-func SetHeaders(ctx context.Context, repo *pb.Repository) (context.Context, error) {
-	repoPath, err := helper.GetPath(repo)
-	if err != nil {
-		return nil, err
-	}
-
-	md := metadata.Pairs(repoPathHeader, repoPath, glRepositoryHeader, repo.GlRepository)
-	newCtx := metadata.NewOutgoingContext(ctx, md)
-	return newCtx, nil
-}
-
-// Proxy calls recvSend until it receives an error. The error is returned
-// to the caller unless it is io.EOF.
-func Proxy(recvSend func() error) (err error) {
-	for err == nil {
-		err = recvSend()
-	}
-
-	if err == io.EOF {
-		err = nil
-	}
-	return err
-}