/*
 *
 * Copyright 2014, Google Inc.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 *     * Redistributions of source code must retain the above copyright
 * notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 *     * Neither the name of Google Inc. nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */

/*
Package transport defines and implements message oriented communication channel
to complete various transactions (e.g., an RPC).
*/
package transport

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"time"

	"golang.org/x/net/context"
	"golang.org/x/net/trace"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

// recvMsg represents the received msg from the transport. All transport
// protocol specific info has been removed.
type recvMsg struct {
	data []byte
	// nil: received some data
	// io.EOF: stream is completed. data is nil.
	// other non-nil error: transport failure. data is nil.
	err error
}

func (recvMsg) isItem() bool {
	return true
}

// All items in an out of a recvBuffer should be the same type.
type item interface {
	isItem() bool
}

// recvBuffer is an unbounded channel of item.
type recvBuffer struct {
	c       chan item
	mu      sync.Mutex
	backlog []item
}

func newRecvBuffer() *recvBuffer {
	b := &recvBuffer{
		c: make(chan item, 1),
	}
	return b
}

func (b *recvBuffer) put(r item) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.backlog = append(b.backlog, r)
	select {
	case b.c <- b.backlog[0]:
		b.backlog = b.backlog[1:]
	default:
	}
}

func (b *recvBuffer) load() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.backlog) > 0 {
		select {
		case b.c <- b.backlog[0]:
			b.backlog = b.backlog[1:]
		default:
		}
	}
}

// get returns the channel that receives an item in the buffer.
//
// Upon receipt of an item, the caller should call load to send another
// item onto the channel if there is any.
func (b *recvBuffer) get() <-chan item {
	return b.c
}

// recvBufferReader implements io.Reader interface to read the data from
// recvBuffer.
type recvBufferReader struct {
	ctx  context.Context
	recv *recvBuffer
	last *bytes.Reader // Stores the remaining data in the previous calls.
	err  error
}

// Read reads the next len(p) bytes from last. If last is drained, it tries to
// read additional data from recv. It blocks if there no additional data available
// in recv. If Read returns any non-nil error, it will continue to return that error.
func (r *recvBufferReader) Read(p []byte) (n int, err error) {
	if r.err != nil {
		return 0, r.err
	}
	defer func() { r.err = err }()
	if r.last != nil && r.last.Len() > 0 {
		// Read remaining data left in last call.
		return r.last.Read(p)
	}
	select {
	case <-r.ctx.Done():
		return 0, ContextErr(r.ctx.Err())
	case i := <-r.recv.get():
		r.recv.load()
		m := i.(*recvMsg)
		if m.err != nil {
			return 0, m.err
		}
		r.last = bytes.NewReader(m.data)
		return r.last.Read(p)
	}
}

type streamState uint8

const (
	streamActive    streamState = iota
	streamWriteDone             // EndStream sent
	streamReadDone              // EndStream received
	streamDone                  // sendDone and recvDone or RSTStreamFrame is sent or received.
)

// Stream represents an RPC in the transport layer.
type Stream struct {
	id uint32
	// nil for client side Stream.
	st ServerTransport
	// ctx is the associated context of the stream.
	ctx    context.Context
	cancel context.CancelFunc
	// method records the associated RPC method of the stream.
	method    string
	buf       *recvBuffer
	dec       io.Reader
	fc        *inFlow
	recvQuota uint32
	// The accumulated inbound quota pending for window update.
	updateQuota uint32
	// The handler to control the window update procedure for both this
	// particular stream and the associated transport.
	windowHandler func(int)

	sendQuotaPool *quotaPool
	// Close headerChan to indicate the end of reception of header metadata.
	headerChan chan struct{}
	// header caches the received header metadata.
	header metadata.MD
	// The key-value map of trailer metadata.
	trailer metadata.MD

	mu sync.RWMutex // guard the following
	// headerOK becomes true from the first header is about to send.
	headerOk bool
	state    streamState
	// true iff headerChan is closed. Used to avoid closing headerChan
	// multiple times.
	headerDone bool
	// the status received from the server.
	statusCode codes.Code
	statusDesc string
}

// Header acquires the key-value pairs of header metadata once it
// is available. It blocks until i) the metadata is ready or ii) there is no
// header metadata or iii) the stream is cancelled/expired.
func (s *Stream) Header() (metadata.MD, error) {
	select {
	case <-s.ctx.Done():
		return nil, ContextErr(s.ctx.Err())
	case <-s.headerChan:
		return s.header.Copy(), nil
	}
}

// Trailer returns the cached trailer metedata. Note that if it is not called
// after the entire stream is done, it could return an empty MD. Client
// side only.
func (s *Stream) Trailer() metadata.MD {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.trailer.Copy()
}

// ServerTransport returns the underlying ServerTransport for the stream.
// The client side stream always returns nil.
func (s *Stream) ServerTransport() ServerTransport {
	return s.st
}

// Context returns the context of the stream.
func (s *Stream) Context() context.Context {
	return s.ctx
}

// TraceContext recreates the context of s with a trace.Trace.
func (s *Stream) TraceContext(tr trace.Trace) {
	s.ctx = trace.NewContext(s.ctx, tr)
}

// Method returns the method for the stream.
func (s *Stream) Method() string {
	return s.method
}

// StatusCode returns statusCode received from the server.
func (s *Stream) StatusCode() codes.Code {
	return s.statusCode
}

// StatusDesc returns statusDesc received from the server.
func (s *Stream) StatusDesc() string {
	return s.statusDesc
}

// ErrIllegalTrailerSet indicates that the trailer has already been set or it
// is too late to do so.
var ErrIllegalTrailerSet = errors.New("transport: trailer has been set")

// SetTrailer sets the trailer metadata which will be sent with the RPC status
// by the server. This can only be called at most once. Server side only.
func (s *Stream) SetTrailer(md metadata.MD) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.trailer != nil {
		return ErrIllegalTrailerSet
	}
	s.trailer = md.Copy()
	return nil
}

func (s *Stream) write(m recvMsg) {
	s.buf.put(&m)
}

// Read reads all the data available for this Stream from the transport and
// passes them into the decoder, which converts them into a gRPC message stream.
// The error is io.EOF when the stream is done or another non-nil error if
// the stream broke.
func (s *Stream) Read(p []byte) (n int, err error) {
	n, err = s.dec.Read(p)
	if err != nil {
		return
	}
	s.windowHandler(n)
	return
}

type key int

// The key to save transport.Stream in the context.
const streamKey = key(0)

// newContextWithStream creates a new context from ctx and attaches stream
// to it.
func newContextWithStream(ctx context.Context, stream *Stream) context.Context {
	return context.WithValue(ctx, streamKey, stream)
}

// StreamFromContext returns the stream saved in ctx.
func StreamFromContext(ctx context.Context) (s *Stream, ok bool) {
	s, ok = ctx.Value(streamKey).(*Stream)
	return
}

// state of transport
type transportState int

const (
	reachable transportState = iota
	unreachable
	closing
)

// NewServerTransport creates a ServerTransport with conn or non-nil error
// if it fails.
func NewServerTransport(protocol string, conn net.Conn, maxStreams uint32, authInfo credentials.AuthInfo) (ServerTransport, error) {
	return newHTTP2Server(conn, maxStreams, authInfo)
}

// ConnectOptions covers all relevant options for dialing a server.
type ConnectOptions struct {
	// UserAgent is the application user agent.
	UserAgent string
	// Dialer specifies how to dial a network address.
	Dialer func(string, time.Duration) (net.Conn, error)
	// AuthOptions stores the credentials required to setup a client connection and/or issue RPCs.
	AuthOptions []credentials.Credentials
	// Timeout specifies the timeout for dialing a client connection.
	Timeout time.Duration
}

// NewClientTransport establishes the transport with the required ConnectOptions
// and returns it to the caller.
func NewClientTransport(target string, opts *ConnectOptions) (ClientTransport, error) {
	return newHTTP2Client(target, opts)
}

// Options provides additional hints and information for message
// transmission.
type Options struct {
	// Indicate whether it is the last piece for this stream.
	Last bool
	// The hint to transport impl whether the data could be buffered for
	// batching write. Transport impl can feel free to ignore it.
	Delay bool
}

// CallHdr carries the information of a particular RPC.
type CallHdr struct {
	Host   string // peer host
	Method string // the operation to perform on the specified host
}

// ClientTransport is the common interface for all gRPC client side transport
// implementations.
type ClientTransport interface {
	// Close tears down this transport. Once it returns, the transport
	// should not be accessed any more. The caller must make sure this
	// is called only once.
	Close() error

	// Write sends the data for the given stream. A nil stream indicates
	// the write is to be performed on the transport as a whole.
	Write(s *Stream, data []byte, opts *Options) error

	// NewStream creates a Stream for an RPC.
	NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error)

	// CloseStream clears the footprint of a stream when the stream is
	// not needed any more. The err indicates the error incurred when
	// CloseStream is called. Must be called when a stream is finished
	// unless the associated transport is closing.
	CloseStream(stream *Stream, err error)

	// Error returns a channel that is closed when some I/O error
	// happens. Typically the caller should have a goroutine to monitor
	// this in order to take action (e.g., close the current transport
	// and create a new one) in error case. It should not return nil
	// once the transport is initiated.
	Error() <-chan struct{}
}

// ServerTransport is the common interface for all gRPC server side transport
// implementations.
type ServerTransport interface {
	// WriteStatus sends the status of a stream to the client.
	WriteStatus(s *Stream, statusCode codes.Code, statusDesc string) error
	// Write sends the data for the given stream.
	Write(s *Stream, data []byte, opts *Options) error
	// WriteHeader sends the header metedata for the given stream.
	WriteHeader(s *Stream, md metadata.MD) error
	// HandleStreams receives incoming streams using the given handler.
	HandleStreams(func(*Stream))
	// Close tears down the transport. Once it is called, the transport
	// should not be accessed any more. All the pending streams and their
	// handlers will be terminated asynchronously.
	Close() error
	// RemoteAddr returns the remote network address.
	RemoteAddr() net.Addr
}

// StreamErrorf creates an StreamError with the specified error code and description.
func StreamErrorf(c codes.Code, format string, a ...interface{}) StreamError {
	return StreamError{
		Code: c,
		Desc: fmt.Sprintf(format, a...),
	}
}

// ConnectionErrorf creates an ConnectionError with the specified error description.
func ConnectionErrorf(format string, a ...interface{}) ConnectionError {
	return ConnectionError{
		Desc: fmt.Sprintf(format, a...),
	}
}

// ConnectionError is an error that results in the termination of the
// entire connection and the retry of all the active streams.
type ConnectionError struct {
	Desc string
}

func (e ConnectionError) Error() string {
	return fmt.Sprintf("connection error: desc = %q", e.Desc)
}

// Define some common ConnectionErrors.
var ErrConnClosing = ConnectionError{Desc: "transport is closing"}

// StreamError is an error that only affects one stream within a connection.
type StreamError struct {
	Code codes.Code
	Desc string
}

func (e StreamError) Error() string {
	return fmt.Sprintf("stream error: code = %d desc = %q", e.Code, e.Desc)
}

// ContextErr converts the error from context package into a StreamError.
func ContextErr(err error) StreamError {
	switch err {
	case context.DeadlineExceeded:
		return StreamErrorf(codes.DeadlineExceeded, "%v", err)
	case context.Canceled:
		return StreamErrorf(codes.Canceled, "%v", err)
	}
	panic(fmt.Sprintf("Unexpected error from context packet: %v", err))
}

// wait blocks until it can receive from ctx.Done, closing, or proceed.
// If it receives from ctx.Done, it returns 0, the StreamError for ctx.Err.
// If it receives from closing, it returns 0, ErrConnClosing.
// If it receives from proceed, it returns the received integer, nil.
func wait(ctx context.Context, closing <-chan struct{}, proceed <-chan int) (int, error) {
	select {
	case <-ctx.Done():
		return 0, ContextErr(ctx.Err())
	case <-closing:
		return 0, ErrConnClosing
	case i := <-proceed:
		return i, nil
	}
}