func TestThatItWorksWithTwoLoggregators(t *testing.T) {
	listener1, dataChan1 := agentlistener.NewAgentListener("localhost:9998", logger)
	go listener1.Start()

	listener2, dataChan2 := agentlistener.NewAgentListener("localhost:9997", logger)
	go listener2.Start()

	loggregatorServers := []string{"localhost:9998", "localhost:9997"}
	hasher := hasher.NewHasher(loggregatorServers)
	rt, err := NewRouter("localhost:3457", hasher, newCfConfig(), logger)
	assert.NoError(t, err)

	go rt.Start(logger)
	time.Sleep(50 * time.Millisecond)

	logEmitter, _ := emitter.NewEmitter("localhost:3457", "ROUTER", "42", "secret", logger)
	logEmitter.Emit("2", "My message")

	receivedData := <-dataChan1
	receivedEnvelope := &logmessage.LogEnvelope{}
	proto.Unmarshal(receivedData, receivedEnvelope)

	assert.Equal(t, receivedEnvelope.GetLogMessage().GetAppId(), "2")
	assert.Equal(t, string(receivedEnvelope.GetLogMessage().GetMessage()), "My message")

	logEmitter.Emit("1", "Another message")

	receivedData = <-dataChan2
	receivedEnvelope = &logmessage.LogEnvelope{}
	proto.Unmarshal(receivedData, receivedEnvelope)

	assert.Equal(t, receivedEnvelope.GetLogMessage().GetAppId(), "1")
	assert.Equal(t, string(receivedEnvelope.GetLogMessage().GetMessage()), "Another message")
}

// Benchmark the Unmarshal() of the raw gogoprotobuf,
// in order to be compared with the codec.
func BenchmarkGoGoProtobufWithoutReflectUnmarshal(b *testing.B) {
	var err error

	messages := generateGoGoProtobufMessages()
	data := make([][]byte, len(messages))

	for j := range messages {
		data[j], err = proto.Marshal(messages[j])
		assert.NoError(b, err)
	}

	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		m1 := new(example.GoGoProtobufTestMessage1)
		assert.NoError(b, proto.Unmarshal(data[0], m1))

		m2 := new(example.GoGoProtobufTestMessage2)
		assert.NoError(b, proto.Unmarshal(data[1], m2))

		m3 := new(example.GoGoProtobufTestMessage3)
		assert.NoError(b, proto.Unmarshal(data[2], m3))

		m4 := new(example.GoGoProtobufTestMessage4)
		assert.NoError(b, proto.Unmarshal(data[3], m4))
	}
}

func (c *Connection) readResponse(response proto.Message) (proto.Message, error) {
	message, ok := <-c.messages
	if !ok {
		return nil, DisconnectedError
	}

	if message.GetType() == warden.Message_Error {
		errorResponse := &warden.ErrorResponse{}
		err := proto.Unmarshal(message.Payload, errorResponse)
		if err != nil {
			return nil, errors.New("error unmarshalling error!")
		}

		return nil, &WardenError{
			Message:   errorResponse.GetMessage(),
			Data:      errorResponse.GetData(),
			Backtrace: errorResponse.GetBacktrace(),
		}
	}

	responseType := warden.TypeForMessage(response)
	if message.GetType() != responseType {
		return nil, errors.New(
			fmt.Sprintf(
				"expected message type %s, got %s\n",
				responseType.String(),
				message.GetType().String(),
			),
		)
	}

	err := proto.Unmarshal(message.GetPayload(), response)

	return response, err
}

func startClientWithSerialization(done chan bool) {
	// with serialization
	s, err := NewSender("localhost:8001")
	if err != nil {
		log.Fatal(err)
	}

	for i := 0; i < 1000; i++ {
		msg := NewMessage(MsgRequireReply+1, nil)
		pa := NewPreAcceptSample() // create a protobuf struct
		pr := new(example.PreAcceptReply)
		// Marshal to bytes
		msg.bytes, err = proto.Marshal(pa)
		if err != nil {
			log.Fatal(err)
		}
		reply, err := s.Send(msg)
		if err != nil {
			log.Fatal(err)
		}
		//Unmarshl the reply bytes
		if err := proto.Unmarshal(reply.Bytes(), pr); err != nil {
			log.Fatal(err)
		}
	}
	done <- true
}

func startServerWithSerialization() {
	withSerializationServerStarted = true
	r := NewReceiver("localhost:8001")
	r.GoStart()

	// with serialization
	go func() {
		for {
			reply := NewPreAcceptReplySample() // create a reply protobuf
			rmsg := NewEmptyMessage()          // create a reply message
			pa := new(example.PreAccept)
			msg := r.Recv()
			rmsg.msgType = msg.msgType
			// Unmarshal the message bytes
			if err := proto.Unmarshal(msg.Bytes(), pa); err != nil {
				log.Fatal(err)
			}
			// Marshal the reply message bytes
			var err error
			rmsg.bytes, err = proto.Marshal(reply)
			if err != nil {
				log.Fatal(err)
			}
			msg.reply <- rmsg
		}
	}()
}

func AssertProtoBufferMessageEquals(t *testing.T, expectedMessage string, actual []byte) {
	receivedMessage := &logmessage.LogMessage{}

	err := proto.Unmarshal(actual, receivedMessage)
	assert.NoError(t, err)
	assert.Equal(t, expectedMessage, string(receivedMessage.GetMessage()))
}

// Decodes the SnapshotRecoveryRequest from a buffer. Returns the number of bytes read and
// any error that occurs.
func (req *SnapshotRecoveryRequest) Decode(r io.Reader) (int, error) {
	data, err := ioutil.ReadAll(r)

	if err != nil {
		return 0, err
	}

	totalBytes := len(data)

	pb := &protobuf.SnapshotRecoveryRequest{}
	if err = proto.Unmarshal(data, pb); err != nil {
		return -1, err
	}

	req.LeaderName = pb.GetLeaderName()
	req.LastIndex = pb.GetLastIndex()
	req.LastTerm = pb.GetLastTerm()
	req.State = pb.GetState()

	req.Peers = make([]*Peer, len(pb.Peers))

	for i, peer := range pb.Peers {
		req.Peers[i] = &Peer{
			Name:             peer.GetName(),
			ConnectionString: peer.GetConnectionString(),
		}
	}

	return totalBytes, nil
}

func (c *InternalConnection) Handle() error {
	defer c.conn.Close()
	// remoteAddr := c.conn.RemoteAddr().String()
	buf := make([]byte, 65536)

	for {
		// c.conn.SetReadDeadline(time.Now().Add(time.Second * 10))
		length, err := binary.ReadUvarint(&proxyByteReader{c.conn})
		if err != nil {
			return err
		}

		packet := buf[0:length]
		var read uint64

		for read < length {
			n, err := c.conn.Read(packet[read:])
			if err != nil {
				return err
			}
			read += uint64(n)
		}

		message := &protobuf.InternalMessage{}
		err = proto.Unmarshal(packet, message)
		if err != nil {
			return err
		}

		c.handler.HandleMessage(message.GetValue(), c)
	}
}

func TestAMarshalTo(t *testing.T) {
	popr := math_rand.New(math_rand.NewSource(time.Now().UnixNano()))
	p := NewPopulatedA(popr, false)
	size := p.Size()
	data := make([]byte, size)
	for i := range data {
		data[i] = byte(popr.Intn(256))
	}
	_, err := p.MarshalTo(data)
	if err != nil {
		panic(err)
	}
	msg := &A{}
	if err := code_google_com_p_gogoprotobuf_proto.Unmarshal(data, msg); err != nil {
		panic(err)
	}
	for i := range data {
		data[i] = byte(popr.Intn(256))
	}
	if err := p.VerboseEqual(msg); err != nil {
		t.Fatalf("%#v !VerboseProto %#v, since %v", msg, p, err)
	}
	if !p.Equal(msg) {
		t.Fatalf("%#v !Proto %#v", msg, p)
	}
}

func (p *ProtobufDecoder) Decode(pack *PipelinePack) (
	packs []*PipelinePack, err error) {

	atomic.AddInt64(&p.processMessageCount, 1)

	var startTime time.Time
	if p.sample {
		startTime = time.Now()
	}

	if err = proto.Unmarshal(pack.MsgBytes, pack.Message); err == nil {
		packs = []*PipelinePack{pack}
	} else {
		atomic.AddInt64(&p.processMessageFailures, 1)
	}

	if p.sample {
		duration := time.Since(startTime).Nanoseconds()
		p.reportLock.Lock()
		p.processMessageDuration += duration
		p.processMessageSamples++
		p.reportLock.Unlock()
	}
	p.sample = 0 == rand.Intn(p.sampleDenominator)
	return
}

func (md *MsgDecoder) DecodePb(m *PbMessage) error {
	msgType, err := md.br.ReadByte()

	if err != nil {
		return err
	}

	m.msgType = uint8(msgType)

	t, ok := registry[m.msgType]

	if !ok {
		panic("unknown type") // TODO error handle
	}

	// since reflect.New() returns a pointer type,
	// so m.pb's underlying type is actually a pointer
	v := reflect.New(t)
	m.pb = v.Interface().(proto.Message)

	var size uint32
	err = binary.Read(md.br, binary.LittleEndian, &size)
	if size == 0 || err != nil { // no need to read and unmarshal
		m.pb = nil
		return err
	}

	bytes := make([]byte, size)
	_, err = io.ReadFull(md.br, bytes)
	if err != nil {
		return err
	}

	return proto.Unmarshal(bytes, m.pb)
}

func TestThatItIgnoresBadMessages(t *testing.T) {
	listener, dataChannel := agentlistener.NewAgentListener("localhost:9996", logger)
	go listener.Start()

	loggregatorServers := []string{"localhost:9996"}
	hasher := hasher.NewHasher(loggregatorServers)
	r, err := NewRouter("localhost:3455", hasher, newCfConfig(), logger)
	assert.NoError(t, err)

	go r.Start(logger)
	time.Sleep(50 * time.Millisecond)

	lc := loggregatorclient.NewLoggregatorClient("localhost:3455", logger, loggregatorclient.DefaultBufferSize)
	lc.Send([]byte("This is poorly formatted"))

	logEmitter, _ := emitter.NewEmitter("localhost:3455", "ROUTER", "42", "secret", logger)
	logEmitter.Emit("my_awesome_app", "Hello World")

	received := <-dataChannel
	receivedEnvelope := &logmessage.LogEnvelope{}
	proto.Unmarshal(received, receivedEnvelope)

	assert.Equal(t, receivedEnvelope.GetLogMessage().GetAppId(), "my_awesome_app")
	assert.Equal(t, string(receivedEnvelope.GetLogMessage().GetMessage()), "Hello World")
}

// Mount makes the contents of the volume visible at the given
// mountpoint. If Mount returns with a nil error, the mount has
// occurred.
func (app *App) Mount(volumeName string, mountpoint string) (*MountInfo, error) {
	// TODO obey `bazil -debug server run`

	var vol *fs.Volume
	var volumeID *fs.VolumeID
	var ready = make(chan error, 1)
	app.mounts.Lock()
	err := app.DB.View(func(tx *bolt.Tx) error {
		bucket := tx.Bucket([]byte(tokens.BucketVolName))
		val := bucket.Get([]byte(volumeName))
		if val == nil {
			return errors.New("volume not found")
		}
		var volConf wire.VolumeConfig
		if err := proto.Unmarshal(val, &volConf); err != nil {
			return err
		}
		var err error
		volumeID, err = fs.NewVolumeID(volConf.VolumeID)
		if err != nil {
			return err
		}
		if _, ok := app.mounts.open[*volumeID]; ok {
			return errors.New("volume already mounted")
		}

		kvstore, err := app.openKV(&volConf.Storage)
		if err != nil {
			return err
		}

		chunkStore := kvchunks.New(kvstore)
		vol, err = fs.Open(app.DB, chunkStore, volumeID)
		if err != nil {
			return err
		}
		mnt := &mountState{
			unmounted: make(chan struct{}),
		}
		go func() {
			defer close(mnt.unmounted)
			ready <- app.serveMount(vol, volumeID, mountpoint)
		}()
		app.mounts.open[*volumeID] = mnt
		return nil
	})
	app.mounts.Unlock()
	if err != nil {
		return nil, err
	}
	err = <-ready
	if err != nil {
		return nil, err
	}
	info := &MountInfo{
		VolumeID: *volumeID,
	}
	return info, nil
}

func (s *SandboxInput) Run(ir pipeline.InputRunner, h pipeline.PluginHelper) (err error) {
	s.sb.InjectMessage(func(payload, payload_type, payload_name string) int {
		pack := <-ir.InChan()
		if err := proto.Unmarshal([]byte(payload), pack.Message); err != nil {
			pack.Recycle()
			return 1
		}
		if s.tz != time.UTC {
			const layout = "2006-01-02T15:04:05.999999999" // remove the incorrect UTC tz info
			t := time.Unix(0, pack.Message.GetTimestamp())
			t = t.In(time.UTC)
			ct, _ := time.ParseInLocation(layout, t.Format(layout), s.tz)
			pack.Message.SetTimestamp(ct.UnixNano())
		}
		ir.Inject(pack)
		atomic.AddInt64(&s.processMessageCount, 1)
		atomic.AddInt64(&s.processMessageBytes, int64(len(payload)))
		return 0
	})

	ticker := ir.Ticker()

	for true {
		retval := s.sb.ProcessMessage(nil)
		if retval <= 0 { // Sandbox is in polling mode
			if retval < 0 {
				atomic.AddInt64(&s.processMessageFailures, 1)
				em := s.sb.LastError()
				if len(em) > 0 {
					ir.LogError(errors.New(em))
				}
			}
			if ticker == nil {
				ir.LogMessage("single run completed")
				break
			}
			select { // block until stop or poll interval
			case <-s.stopChan:
			case <-ticker:
			}
		} else { // Sandbox is shutting down
			em := s.sb.LastError()
			if !strings.HasSuffix(em, "shutting down") {
				ir.LogError(errors.New(em))
			}
			break
		}
	}

	s.reportLock.Lock()
	if s.sbc.PreserveData {
		err = s.sb.Destroy(s.preservationFile)
	} else {
		err = s.sb.Destroy("")
	}
	s.sb = nil
	s.reportLock.Unlock()
	return
}

func extractLogEnvelope(data []byte) (*logmessage.LogEnvelope, error) {
	logEnvelope := new(logmessage.LogEnvelope)
	err := proto.Unmarshal(data, logEnvelope)
	if err != nil {
		return logEnvelope, err
	}
	return logEnvelope, nil
}

func extractLogEnvelope(data *[]byte) *logmessage.LogEnvelope {
	receivedEnvelope := &logmessage.LogEnvelope{}

	err := proto.Unmarshal(*data, receivedEnvelope)
	Ω(err).ShouldNot(HaveOccurred())

	return receivedEnvelope
}

func parseProtoBufMessageString(actual []byte) string {
	receivedMessage := &logmessage.LogMessage{}
	err := proto.Unmarshal(actual, receivedMessage)
	if err != nil {
		Fail(err.Error())
	}
	return string(receivedMessage.GetMessage())
}

func (d *dir) unmarshalDirent(buf []byte) (*wire.Dirent, error) {
	var de wire.Dirent
	err := proto.Unmarshal(buf, &de)
	if err != nil {
		return nil, err
	}
	return &de, nil
}

func (e *loggingEmitter) Emit(data []byte) (err error) {
	envelope := new(events.Envelope)
	err = proto.Unmarshal(data, envelope)
	if err != nil {
		return
	}

	log.Printf("Emitting %s\n", proto.CompactTextString(envelope))
	return
}

// GetProto fetches the value at the specified key and unmarshals it
// using a protobuf decoder. See comments for GetI for details on
// return values.
func GetProto(db DB, key engine.Key, msg gogoproto.Message) (bool, proto.Timestamp, error) {
	ok, value, err := getInternal(db, key)
	if !ok || err != nil {
		return false, proto.Timestamp{}, err
	}
	if err := gogoproto.Unmarshal(value.Bytes, msg); err != nil {
		return true, value.Timestamp, err
	}
	return true, value.Timestamp, nil
}