func (d *resumableStreamDecoder) changeState(target resumableStreamDecoderState) {
	if d.state == queueingRetryable && d.state != target {
		d.bytesBetweenResumeTokens = 0
	}
	d.state = target
	if d.stateWitness != nil {
		d.stateWitness(target)
	}
}
changeState fulfills state transition for resumableStateDecoder.
package spanner
import (
	"bytes"
	"io"
	"log"
	"sync/atomic"
	"time"
	"cloud.google.com/go/internal/protostruct"
	proto "github.com/golang/protobuf/proto"
	proto3 "github.com/golang/protobuf/ptypes/struct"
	"golang.org/x/net/context"
	"google.golang.org/api/iterator"
	sppb "google.golang.org/genproto/googleapis/spanner/v1"
	"google.golang.org/grpc/codes"
)
type streamingReceiver interface {
	Recv() (*sppb.PartialResultSet, error)
}
func errEarlyReadEnd() error {
	return spannerErrorf(codes.FailedPrecondition, "read completed with active stream")
}
func stream(ctx context.Context, rpc func(ct context.Context, resumeToken []byte) (streamingReceiver, error), setTimestamp func(time.Time), release func(error)) *RowIterator {
	ctx, cancel := context.WithCancel(ctx)
	ctx = traceStartSpan(ctx, "cloud.google.com/go/spanner.RowIterator")
	return &RowIterator{
		streamd:      newResumableStreamDecoder(ctx, rpc),
		rowd:         &partialResultSetDecoder{},
		setTimestamp: setTimestamp,
		release:      release,
		cancel:       cancel,
	}
}
type RowIterator struct {
	QueryPlan *sppb.QueryPlan
	QueryStats map[string]interface{}
	streamd      *resumableStreamDecoder
	rowd         *partialResultSetDecoder
	setTimestamp func(time.Time)
	release      func(error)
	cancel       func()
	err          error
	rows         []*Row
}
func (r *RowIterator) Next() (*Row, error) {
	if r.err != nil {
		return nil, r.err
	}
	for len(r.rows) == 0 && r.streamd.next() {
		prs := r.streamd.get()
		if prs.Stats != nil {
			r.QueryPlan = prs.Stats.QueryPlan
			r.QueryStats = protostruct.DecodeToMap(prs.Stats.QueryStats)
		}
		r.rows, r.err = r.rowd.add(prs)
		if r.err != nil {
			return nil, r.err
		}
		if !r.rowd.ts.IsZero() && r.setTimestamp != nil {
			r.setTimestamp(r.rowd.ts)
			r.setTimestamp = nil
		}
	}
	if len(r.rows) > 0 {
		row := r.rows[0]
		r.rows = r.rows[1:]
		return row, nil
	}
	if err := r.streamd.lastErr(); err != nil {
		r.err = toSpannerError(err)
	} else if !r.rowd.done() {
		r.err = errEarlyReadEnd()
	} else {
		r.err = iterator.Done
	}
	return nil, r.err
}
func (r *RowIterator) Do(f func(r *Row) error) error {
	defer r.Stop()
	for {
		row, err := r.Next()
		switch err {
		case iterator.Done:
			return nil
		case nil:
			if err = f(row); err != nil {
				return err
			}
		default:
			return err
		}
	}
}
func (r *RowIterator) Stop() {
	if r.streamd != nil {
		defer traceEndSpan(r.streamd.ctx, r.err)
	}
	if r.cancel != nil {
		r.cancel()
	}
	if r.release != nil {
		r.release(r.err)
		if r.err == nil {
			r.err = spannerErrorf(codes.FailedPrecondition, "Next called after Stop")
		}
		r.release = nil
	}
}
type partialResultQueue struct {
	q     []*sppb.PartialResultSet
	first int
	last  int
	n     int  
}
func (q *partialResultQueue) empty() bool {
	return q.n == 0
}
func errEmptyQueue() error {
	return spannerErrorf(codes.OutOfRange, "empty partialResultQueue")
}
func (q *partialResultQueue) peekLast() (*sppb.PartialResultSet, error) {
	if q.empty() {
		return nil, errEmptyQueue()
	}
	return q.q[(q.last+cap(q.q)-1)%cap(q.q)], nil
}
func (q *partialResultQueue) push(r *sppb.PartialResultSet) {
	if q.q == nil {
		q.q = make([]*sppb.PartialResultSet, 8  )
	}
	if q.n == cap(q.q) {
		buf := make([]*sppb.PartialResultSet, cap(q.q)*2)
		for i := 0; i < q.n; i++ {
			buf[i] = q.q[(q.first+i)%cap(q.q)]
		}
		q.q = buf
		q.first = 0
		q.last = q.n
	}
	q.q[q.last] = r
	q.last = (q.last + 1) % cap(q.q)
	q.n++
}
func (q *partialResultQueue) pop() *sppb.PartialResultSet {
	if q.n == 0 {
		return nil
	}
	r := q.q[q.first]
	q.q[q.first] = nil
	q.first = (q.first + 1) % cap(q.q)
	q.n--
	return r
}
func (q *partialResultQueue) clear() {
	*q = partialResultQueue{}
}
func (q *partialResultQueue) dump() []*sppb.PartialResultSet {
	var dq []*sppb.PartialResultSet
	for i := q.first; len(dq) < q.n; i = (i + 1) % cap(q.q) {
		dq = append(dq, q.q[i])
	}
	return dq
}
type resumableStreamDecoderState int
const (
	unConnected         resumableStreamDecoderState = iota  
	queueingRetryable                                       
	queueingUnretryable                                     
	aborted                                                 
	finished                                                
)
type resumableStreamDecoder struct {
	state resumableStreamDecoderState
	stateWitness func(resumableStreamDecoderState)
	ctx context.Context
	rpc func(ctx context.Context, restartToken []byte) (streamingReceiver, error)
	stream streamingReceiver
	q partialResultQueue
	bytesBetweenResumeTokens int32
	maxBytesBetweenResumeTokens int32
	np *sppb.PartialResultSet
	resumeToken []byte
	retryCount int
	err error
	backoff exponentialBackoff
}
func newResumableStreamDecoder(ctx context.Context, rpc func(ct context.Context, restartToken []byte) (streamingReceiver, error)) *resumableStreamDecoder {
	return &resumableStreamDecoder{
		ctx: ctx,
		rpc: rpc,
		maxBytesBetweenResumeTokens: atomic.LoadInt32(&maxBytesBetweenResumeTokens),
		backoff:                     defaultBackoff,
	}
}