func (s *BulkPullBlocksSyncer) ReadNext(r io.Reader) (bool, error) {
	blk, err := readBlock(r)
	if err != nil {
		if err == block.ErrNotABlock {
			return true, nil
		}
		return false, err
	}
	s.blocks = append(s.blocks, blk)
	if len(s.blocks) >= syncCacheSize {
		s.Flush()
	}
	return false, nil
}
ReadNext implements the Syncer interface.
package node
import (
	"bufio"
	"io"
	"math"
	"net"
	"time"
	"github.com/alexbakker/gonano/nano"
	"github.com/alexbakker/gonano/nano/block"
	"github.com/alexbakker/gonano/nano/internal/util"
	"github.com/alexbakker/gonano/nano/node/proto"
)
const (
	syncTimeout   = time.Second * 2
	syncCacheSize = 10000
)
type Syncer interface {
	Flush()
	ReadNext(r io.Reader) (done bool, err error)
	WriteNext(p *proto.Proto, w io.Writer) (done bool, err error)
}
type (
	FrontierSyncerFunc       func(*block.Frontier)
	BulkPullSyncerFunc       func([]block.Block)
	BulkPullBlocksSyncerFunc func([]block.Block)
)
type FrontierSyncer struct {
	cb FrontierSyncerFunc
}
type BulkPullSyncer struct {
	blocks     []block.Block
	addrs      []nano.Address
	writeIndex int
	readIndex  int
	cb         BulkPullSyncerFunc
}
type BulkPullBlocksSyncer struct {
	blocks []block.Block
	mode   proto.BulkPullMode
	cb     BulkPullBlocksSyncerFunc
}
func NewFrontierSyncer(cb FrontierSyncerFunc) *FrontierSyncer {
	return &FrontierSyncer{cb: cb}
}
func NewBulkPullSyncer(cb BulkPullSyncerFunc, frontiers map[nano.Address]block.Hash) *BulkPullSyncer {
	addrs := make([]nano.Address, 0, len(frontiers))
	for addr := range frontiers {
		addrs = append(addrs, addr)
	}
	return &BulkPullSyncer{
		cb:     cb,
		addrs:  addrs,
		blocks: make([]block.Block, 0, syncCacheSize),
	}
}
func NewBulkPullBlocksSyncer(cb BulkPullBlocksSyncerFunc) *BulkPullBlocksSyncer {
	return &BulkPullBlocksSyncer{
		cb:     cb,
		mode:   proto.BulkPullModeList,
		blocks: make([]block.Block, 0, syncCacheSize),
	}
}
func Sync(syncer Syncer, p *proto.Proto, peer *Peer) error {
	conn, err := initSync(peer)
	if err != nil {
		return err
	}
	defer conn.Close()
	errs := make(chan error)
	cancel := make(chan struct{})
	go func(errs chan<- error, cancel <-chan struct{}) {
		for {
			select {
			case <-cancel:
				errs <- nil
				return
			default:
			}
			if err := conn.SetWriteDeadline(time.Now().Add(syncTimeout)); err != nil {
				errs <- err
				return
			}
			done, err := syncer.WriteNext(p, conn)
			if err != nil {
				errs <- err
				return
			}
			if done {
				errs <- nil
				return
			}
		}
	}(errs, cancel)
	go func(errs chan<- error, cancel <-chan struct{}) {
		reader := bufio.NewReader(conn)
		for {
			select {
			case <-cancel:
				errs <- nil
				return
			default:
			}
			if err := conn.SetReadDeadline(time.Now().Add(syncTimeout)); err != nil {
				errs <- err
				return
			}
			done, err := syncer.ReadNext(reader)
			if err != nil {
				errs <- err
				return
			}
			if done {
				errs <- nil
				return
			}
		}
	}(errs, cancel)
	var res error
	for i := 0; i < 2; {
		select {
		case err := <-errs:
			if err != nil {
				res = err
				close(cancel)
			}
			i++
		}
	}
	if res != nil {
		return err
	}
	syncer.Flush()
	return nil
}
func (s *FrontierSyncer) ReadNext(r io.Reader) (bool, error) {
	var buf [block.FrontierSize]byte
	if _, err := io.ReadFull(r, buf[:]); err != nil {
		return false, err
	}
	var frontier block.Frontier
	if err := frontier.UnmarshalBinary(buf[:]); err != nil {
		return false, err
	}
	if frontier.IsZero() {
		return true, nil
	}
	s.cb(&frontier)
	return false, nil
}
func (s *FrontierSyncer) WriteNext(p *proto.Proto, w io.Writer) (done bool, err error) {
	packet := proto.FrontierReqPacket{
		StartAddress: nano.Address{},
		Age:          math.MaxUint32,
		Count:        math.MaxUint32,
	}
	return true, writePacket(w, p, &packet)
}
func (s *FrontierSyncer) Flush() {
}
func (s *BulkPullSyncer) ReadNext(r io.Reader) (bool, error) {
	blk, err := readBlock(r)
	if err != nil {
		if err == block.ErrNotABlock {
			s.readIndex++
			if s.readIndex >= len(s.addrs) {
				return true, nil
			}
			return false, nil
		}
		return false, err
	}
	s.blocks = append(s.blocks, blk)
	if len(s.blocks) >= syncCacheSize {
		s.Flush()
	}
	return false, nil
}
func (s *BulkPullSyncer) WriteNext(p *proto.Proto, w io.Writer) (done bool, err error) {
	if s.writeIndex < len(s.addrs) {
		packet := proto.BulkPullPacket{
			Address: s.addrs[s.writeIndex],
		}
		s.writeIndex++
		return false, writePacket(w, p, &packet)
	}
	return true, nil
}
func (s *BulkPullSyncer) Flush() {
	if len(s.blocks) > 0 {
		s.cb(s.blocks)
		s.blocks = s.blocks[:0]
	}
}