grpcutil

Documentation

Index

• Constants
• Variables
• func Chunk(data []byte, chunkSize int) [][]byte
• func ChunkReader(r io.Reader, f func([]byte) error) (int, error)
• func GetBuffer() []byte
• func NewStreamingBytesReader(streamingBytesClient StreamingBytesClient, cancel context.CancelFunc) io.ReadCloser
• func NewStreamingBytesWriter(streamingBytesServer StreamingBytesServer) io.Writer
• func PutBuffer(buf []byte)
• func ScrubGRPC(err error) error
• func WriteFromStreamingBytesClient(streamingBytesClient StreamingBytesClient, writer io.Writer) error
• func WriteToStreamingBytesServer(reader io.Reader, streamingBytesServer StreamingBytesServer) error
• type BufPool
  • func NewBufPool(size int) *BufPool
  • func (b *BufPool) GetBuffer() []byte
  • func (b *BufPool) PutBuffer(buf []byte)
• type ChunkWriteCloser
  • func NewChunkWriteCloser(bufPool *BufPool, f func(chunk []byte) error) *ChunkWriteCloser
  • func (w *ChunkWriteCloser) Close() error
  • func (w *ChunkWriteCloser) Write(data []byte) (int, error)
• type Dialer
  • func NewDialer(opts ...grpc.DialOption) Dialer
• type PachdAddress
  • func ParsePachdAddress(value string) (*PachdAddress, error)
  • func (p *PachdAddress) Hostname() string
  • func (p *PachdAddress) IsLoopback() bool
  • func (p *PachdAddress) IsUnusualPort() bool
  • func (p *PachdAddress) Qualified() string
• type ReaderWrapper
  • func (r ReaderWrapper) Read(p []byte) (int, error)
• type Server
  • func NewServer(ctx context.Context, publicPortTLSAllowed bool) (*Server, error)
  • func (s *Server) ListenTCP(host string, port uint16) (net.Listener, error)
  • func (s *Server) Wait() error
• type StreamingBytesClient
• type StreamingBytesServer

Constants

const (
	// DefaultPachdNodePort is the pachd kubernetes service's default
	// NodePort.Port setting
	DefaultPachdNodePort = 30650

	// DefaultPachdPort is the pachd kubernetes service's default
	// Port (often used with Pachyderm ELBs)
	DefaultPachdPort = 650
)

Variables

var (
	// ErrNoPachdAddress is returned by ParsePachdAddress when the input is an
	// empty string
	ErrNoPachdAddress = errors.New("no pachd address specified")
	// DefaultPachdAddress is the default PachdAddress that should be used
	// if none is otherwise specified. It's a loopback that should rely on
	// port forwarding.
	DefaultPachdAddress = PachdAddress{
		Secured: false,
		Host:    "0.0.0.0",
		Port:    DefaultPachdNodePort,
	}
)

var (
	// MaxMsgSize is used to define the GRPC frame size
	MaxMsgSize = 20 * 1024 * 1024
)

Functions

func Chunk

func Chunk(data []byte, chunkSize int) [][]byte

Chunk splits a piece of data up, this is useful for splitting up data that's bigger than MaxMsgSize

func ChunkReader

func ChunkReader(r io.Reader, f func([]byte) error) (int, error)

ChunkReader splits a reader into reasonably sized chunks for the purpose of transmitting the chunks over gRPC. For each chunk, it calls the given function.

func GetBuffer

func GetBuffer() []byte

GetBuffer returns a buffer. The buffer may or may not be freshly allocated, and it may or may not be zero-ed.

func NewStreamingBytesReader

func NewStreamingBytesReader(streamingBytesClient StreamingBytesClient, cancel context.CancelFunc) io.ReadCloser

NewStreamingBytesReader returns an io.Reader for a StreamingBytesClient.

func NewStreamingBytesWriter

func NewStreamingBytesWriter(streamingBytesServer StreamingBytesServer) io.Writer

NewStreamingBytesWriter returns an io.Writer for a StreamingBytesServer.

func PutBuffer

func PutBuffer(buf []byte)

PutBuffer returns the buffer to the pool.

func ScrubGRPC

func ScrubGRPC(err error) error

ScrubGRPC removes GRPC error code information from 'err' if it came from GRPC (and returns it unchanged otherwise)

func WriteFromStreamingBytesClient

func WriteFromStreamingBytesClient(streamingBytesClient StreamingBytesClient, writer io.Writer) error

WriteFromStreamingBytesClient writes from the StreamingBytesClient to the io.Writer.

func WriteToStreamingBytesServer

func WriteToStreamingBytesServer(reader io.Reader, streamingBytesServer StreamingBytesServer) error

WriteToStreamingBytesServer writes the data from the io.Reader to the StreamingBytesServer.

Types

type BufPool

type BufPool struct {
	sync.Pool
}

BufPool is a wrapper around sync.Pool that makes it a little nicer to use for []byte by doing the casting for you and defining the `New` function.

func NewBufPool

func NewBufPool(size int) *BufPool

NewBufPool creates a new BufPool that returns buffers of the given size.

func (*BufPool) GetBuffer

func (b *BufPool) GetBuffer() []byte

GetBuffer returns a buffer. The buffer may or may not be freshly allocated, and it may or may not be zero-ed.

func (*BufPool) PutBuffer

func (b *BufPool) PutBuffer(buf []byte)

PutBuffer returns the buffer to the pool.

type ChunkWriteCloser

type ChunkWriteCloser struct {
	// contains filtered or unexported fields
}

ChunkWriteCloser is a utility for buffering writes into buffers obtained from a buffer pool. The ChunkWriteCloser will buffer up to the capacity of a buffer obtained from a buffer pool, then execute a callback that will receive the buffered data. The ChunkWriteCloser will get a new buffer from the pool for subsequent writes, so it is expected that the callback will return the buffer to the pool.

func NewChunkWriteCloser

func NewChunkWriteCloser(bufPool *BufPool, f func(chunk []byte) error) *ChunkWriteCloser

NewChunkWriteCloser creates a new ChunkWriteCloser.

func (*ChunkWriteCloser) Close

func (w *ChunkWriteCloser) Close() error

Close closes the writer.

func (*ChunkWriteCloser) Write

func (w *ChunkWriteCloser) Write(data []byte) (int, error)

Write performs a write.

type Dialer

type Dialer interface {
	Dial(address string) (*grpc.ClientConn, error)
	CloseConns() error
}

Dialer defines a grpc.ClientConn connection dialer.

func NewDialer

func NewDialer(opts ...grpc.DialOption) Dialer

NewDialer creates a Dialer.

type PachdAddress

type PachdAddress struct {
	// Secured specifies whether grpcs should be used
	Secured bool
	// Host specifies the pachd address host without the port
	Host string
	// Port specifies the pachd port
	Port uint16
}

PachdAddress represents a parsed pachd address value

func ParsePachdAddress

func ParsePachdAddress(value string) (*PachdAddress, error)

ParsePachdAddress parses a string into a pachd address, or returns an error if it's invalid

func (*PachdAddress) Hostname

func (p *PachdAddress) Hostname() string

Hostname returns the host:port combination of the pachd address, without the scheme

func (*PachdAddress) IsLoopback

func (p *PachdAddress) IsLoopback() bool

IsLoopback returns whether the pachd address is referencing the loopback hostname

func (*PachdAddress) IsUnusualPort

func (p *PachdAddress) IsUnusualPort() bool

IsUnusualPort returns true if the pachd address port is not one of the usual values

func (*PachdAddress) Qualified

func (p *PachdAddress) Qualified() string

Qualified returns the "fully qualified" address, including the scheme

type ReaderWrapper

type ReaderWrapper struct {
	Reader io.Reader
}

ReaderWrapper wraps a reader for the following reason: Go's io.CopyBuffer has an annoying optimization wherein if the reader has the WriteTo function defined, it doesn't actually use the given buffer. As a result, we might write a large chunk to the gRPC streaming server even though we intend to use a small buffer. Therefore we wrap readers in this wrapper so that only Read is defined.

func (ReaderWrapper) Read

func (r ReaderWrapper) Read(p []byte) (int, error)

type Server

type Server struct {
	Server *grpc.Server
	// contains filtered or unexported fields
}

Server is a convenience wrapper to gRPC servers that simplifies their setup and execution

func NewServer

func NewServer(ctx context.Context, publicPortTLSAllowed bool) (*Server, error)

NewServer creates a new gRPC server, but does not start serving yet.

If 'publicPortTLSAllowed' is set, grpcutil may enable TLS. This should be set for public ports that serve GRPC services to 3rd party clients. If set, the criterion for actually serving over TLS is: if a signed TLS cert and corresponding private key in 'TLSVolumePath', this will serve GRPC traffic over TLS. If either are missing this will serve GRPC traffic over unencrypted HTTP,

func (*Server) ListenTCP

func (s *Server) ListenTCP(host string, port uint16) (net.Listener, error)

ListenTCP causes the gRPC server to listen on a given TCP host and port

func (*Server) Wait

func (s *Server) Wait() error

Wait causes the gRPC server to wait until it finishes, returning any errors that happened

type StreamingBytesClient

type StreamingBytesClient interface {
	Recv() (*types.BytesValue, error)
}

StreamingBytesClient represents a client for an rpc method of the form:

rpc Foo(Bar) returns (stream google.protobuf.BytesValue) {}

type StreamingBytesServer

type StreamingBytesServer interface {
	Send(bytesValue *types.BytesValue) error
}

StreamingBytesServer represents a server for an rpc method of the form:

rpc Foo(Bar) returns (stream google.protobuf.BytesValue) {}