internal

Documentation

Index

• type ReadHandler
• type Server
  • func NewServer(gsrv *grpc.Server, readHandler ReadHandler, writeHandler WriteHandler) (*Server, error)
• type WriteHandler

Examples

• NewServer

Types

type ReadHandler

type ReadHandler interface {
	// GetReader provides an io.ReaderAt, which will not be retained by the Server after the pb.ReadRequest.
	GetReader(ctx context.Context, name string) (io.ReaderAt, error)
	// Close does not have to do anything, but is here for if the io.ReaderAt wants to call Close().
	Close(ctx context.Context, name string) error
}

ReadHandler reads from the Bytestream. Note: error returns must return an instance of grpc.rpcError unless otherwise handled in grpc-go/rpc_util.go. http://google.golang.org/grpc provides Errorf(code, fmt, ...) to create instances of grpc.rpcError. Note: Cancelling the context will abort the stream ("drop the connection"). Consider returning a non-nil error instead.

type Server

type Server struct {

	// AllowOverwrite controls Server behavior when a WriteRequest with finish_write = true is followed by another WriteRequest.
	AllowOverwrite bool

	// Bytestream allows a WriteRequest to omit the resource name, in which case it will be appended to the last WriteRequest.
	LastWrittenResource string
	// contains filtered or unexported fields
}

Server wraps the RPCs in pb. Use bytestream.NewServer() to create a Server.

func NewServer

func NewServer(gsrv *grpc.Server, readHandler ReadHandler, writeHandler WriteHandler) (*Server, error)

NewServer creates a new bytestream.Server using gRPC. gsrv is the *grpc.Server this bytestream.Server will listen on. readHandler handles any incoming pb.ReadRequest or nil which means all pb.ReadRequests will be rejected. writeHandler handles any incoming pb.WriteRequest or nil which means all pb.WriteRequests will be rejected. readHandler and writeHandler cannot both be nil.

Example

package main

import (
	"bytes"
	"context"
	"io"
	"log"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
)

type ExampleReadHandler struct {
	buf  []byte
	name string // In this example, the service can handle one name only.
}

func (mr *ExampleReadHandler) GetReader(ctx context.Context, name string) (io.ReaderAt, error) {
	if mr.name == "" {
		mr.name = name
		log.Printf("read from name: %q", name)
	} else if mr.name != name {
		return nil, grpc.Errorf(codes.NotFound, "reader has name %q, name %q not allowed", mr.name, name)
	}
	return bytes.NewReader(mr.buf), nil
}

// Close can be a no-op.
func (mr *ExampleReadHandler) Close(ctx context.Context, name string) error {
	return nil
}

type ExampleWriteHandler struct {
	buf  bytes.Buffer // bytes.Buffer implements io.Writer
	name string       // In this example, the service can handle one name only.
}

// Handle writes to a given name.
func (mw *ExampleWriteHandler) GetWriter(ctx context.Context, name string, initOffset int64) (io.Writer, error) {
	if mw.name == "" {
		mw.name = name
		log.Printf("write to name: %q", name)
	} else if mw.name != name {
		return nil, grpc.Errorf(codes.NotFound, "reader has name %q, name=%q not allowed", mw.name, name)
	}
	// TODO: initOffset is ignored.
	return &mw.buf, nil
}

// Close can be a no-op.
func (mw *ExampleWriteHandler) Close(ctx context.Context, name string) error {
	return nil
}

func main() {
	reader := &ExampleReadHandler{
		buf:  []byte("Hello World!"),
		name: "foo",
	}
	writer := &ExampleWriteHandler{}
	gsrv := grpc.NewServer()
	bytestreamServer, err := NewServer(gsrv, reader, writer)
	if err != nil {
		log.Printf("NewServer: %v", err)
		return
	}

	// Start accepting incoming connections.
	// See gRPC docs and newGRPCServer in google.golang.org/api/transport/bytestream/client_test.go.
	_ = bytestreamServer
}

type WriteHandler

type WriteHandler interface {
	// GetWriter provides an io.Writer that is ready to write at initOffset.
	// The io.Writer will not be retained by the Server after the pb.WriteRequest.
	GetWriter(ctx context.Context, name string, initOffset int64) (io.Writer, error)
	// Close does not have to do anything, but is related to Server.AllowOverwrite. Or if the io.Writer simply wants a Close() call.
	// Close is called when the server receives a pb.WriteRequest with finish_write = true.
	// If Server.AllowOverwrite == true then Close() followed by GetWriter() for the same name indicates the name is being overwritten, even if the initOffset is different.
	Close(ctx context.Context, name string) error
}

WriteHandler handles writes from the Bytestream. For example: Note: error returns must return an instance of grpc.rpcError unless otherwise handled in grpc-go/rpc_util.go. grpc-go/rpc_util.go provides the helper func Errorf(code, fmt, ...) to create instances of grpc.rpcError. Note: Cancelling the context will abort the stream ("drop the connection"). Consider returning a non-nil error instead.