func (s *blockWriteStream) ackPackets() {
	reader := bufio.NewReader(s.conn)
	for {
		p, ok := <-s.packets
		if !ok {
			return
		}
		ack := &hdfs.PipelineAckProto{}
		err := readPrefixedMessage(reader, ack)
		if err != nil {
			s.ackError = err
			break
		}
		seqno := int(ack.GetSeqno())
		for i, status := range ack.GetStatus() {
			if status != hdfs.Status_SUCCESS {
				s.ackError = ackError{status: status, seqno: seqno, pipelineIndex: i}
				break
			}
		}
		if seqno != p.seqno {
			s.ackError = ErrInvalidSeqno
			break
		}
	}
	for _ = range s.packets {
	}
}
ackPackets is meant to run in the background, reading acks and setting
ackError if one fails.
package rpc
import (
	"bufio"
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"math"
	hdfs "github.com/colinmarc/hdfs/protocol/hadoop_hdfs"
	"github.com/golang/protobuf/proto"
)
const (
	outboundPacketSize = 65536
	outboundChunkSize  = 512
	maxPacketsInQueue  = 5
)
type blockWriteStream struct {
	block *hdfs.LocatedBlockProto
	conn   io.ReadWriter
	buf    bytes.Buffer
	offset int64
	closed bool
	packets chan outboundPacket
	seqno   int
	ackError        error
	acksDone        chan struct{}
	lastPacketSeqno int
}
type outboundPacket struct {
	seqno     int
	offset    int64
	last      bool
	checksums []byte
	data      []byte
}
type ackError struct {
	pipelineIndex int
	seqno         int
	status        hdfs.Status
}
func (ae ackError) Error() string {
	return fmt.Sprintf("Ack error from datanode: %s", ae.status.String())
}
var ErrInvalidSeqno = errors.New("Invalid ack sequence number")
func newBlockWriteStream(conn io.ReadWriter) *blockWriteStream {
	s := &blockWriteStream{
		conn:     conn,
		offset:   0,
		seqno:    1,
		packets:  make(chan outboundPacket, maxPacketsInQueue),
		acksDone: make(chan struct{}),
	}
	go func() {
		s.ackPackets()
		close(s.acksDone)
	}()
	return s
}
func (s *blockWriteStream) Write(b []byte) (int, error) {
	if s.closed {
		return 0, io.ErrClosedPipe
	}
	if s.ackError != nil {
		return 0, s.ackError
	}
	n, _ := s.buf.Write(b)
	err := s.flush(false)
	return n, err
}
func (s *blockWriteStream) finish() error {
	if s.closed {
		return nil
	}
	s.closed = true
	if s.ackError != nil {
		return s.ackError
	}
	err := s.flush(true)
	if err != nil {
		return err
	}
	lastPacket := outboundPacket{
		seqno:     s.seqno,
		offset:    s.offset,
		last:      true,
		checksums: []byte{},
		data:      []byte{},
	}
	s.packets <- lastPacket
	err = s.writePacket(lastPacket)
	if err != nil {
		return err
	}
	close(s.packets)
	<-s.acksDone
	if s.ackError != nil {
		return s.ackError
	}
	return nil
}
func (s *blockWriteStream) flush(force bool) error {
	for s.buf.Len() > 0 && (force || s.buf.Len() >= outboundPacketSize) {
		packet := s.makePacket()
		s.packets <- packet
		s.offset += int64(len(packet.data))
		s.seqno++
		err := s.writePacket(packet)
		if err != nil {
			return err
		}
	}
	return nil
}
func (s *blockWriteStream) makePacket() outboundPacket {
	packetLength := outboundPacketSize
	if s.buf.Len() < outboundPacketSize {
		packetLength = s.buf.Len()
	}
	numChunks := int(math.Ceil(float64(packetLength) / float64(outboundChunkSize)))
	packet := outboundPacket{
		seqno:     s.seqno,
		offset:    s.offset,
		last:      false,
		checksums: make([]byte, numChunks*4),
		data:      make([]byte, packetLength),
	}
	io.ReadFull(&s.buf, packet.data)
	for i := 0; i < numChunks; i++ {
		chunkOff := i * outboundChunkSize
		chunkEnd := chunkOff + outboundChunkSize
		if chunkEnd >= len(packet.data) {
			chunkEnd = len(packet.data)
		}
		checksum := crc32.Checksum(packet.data[chunkOff:chunkEnd], crc32.IEEETable)
		binary.BigEndian.PutUint32(packet.checksums[i*4:], checksum)
	}
	return packet
}