func (e *LoggregatorEmitter) EmitLogMessage(logMessage *logmessage.LogMessage) {
	messages := splitMessage(string(logMessage.GetMessage()))

	for _, message := range messages {
		if isEmpty(message) {
			continue
		}

		if len(message) > MAX_MESSAGE_BYTE_SIZE {
			logMessage.Message = append([]byte(message)[0:TRUNCATED_OFFSET], TRUNCATED_BYTES...)
		} else {
			logMessage.Message = []byte(message)
		}
		if e.sharedSecret == "" {
			marshalledLogMessage, err := proto.Marshal(logMessage)
			if err != nil {
				e.logger.Errorf("Error marshalling message: %s", err)
				return
			}
			e.LoggregatorClient.Send(marshalledLogMessage)
		} else {
			logEnvelope, err := e.newLogEnvelope(*logMessage.AppId, logMessage)
			if err != nil {
				e.logger.Errorf("Error creating envelope: %s", err)
				return
			}
			marshalledLogEnvelope, err := proto.Marshal(logEnvelope)
			if err != nil {
				e.logger.Errorf("Error marshalling envelope: %s", err)
				return
			}
			e.LoggregatorClient.Send(marshalledLogEnvelope)
		}
	}
}

func TestNoMergeExtensionMerge(t *testing.T) {
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	bigm := test.NewPopulatedMyExtendable(r, true)
	m := test.NewPopulatedNinOptNative(r, true)
	err := proto.SetExtension(bigm, test.E_FieldB, m)
	if err != nil {
		panic(err)
	}
	data, err := proto.Marshal(bigm)
	if err != nil {
		panic(err)
	}
	key := uint32(101)<<3 | uint32(2)
	data2 := make([]byte, 10)
	n := binary.PutUvarint(data2, uint64(key))
	data2 = data2[:n]
	data = append(data, data2...)
	data4, err := proto.Marshal(test.NewPopulatedNinOptNative(r, true))
	if err != nil {
		panic(err)
	}
	data3 := make([]byte, 10)
	n = binary.PutUvarint(data3, uint64(len(data4)))
	data3 = data3[:n]
	data = append(data, data3...)
	data = append(data, data4...)
	err = fieldpath.NoMerge(data, test.ThetestDescription(), "test", "MyExtendable")
	if err == nil || !strings.Contains(err.Error(), "requires merging") {
		t.Fatalf("should require merging")
	}
}

// Mkdir takes a snapshot of this volume and records it under the
// given name.
func (d *listSnaps) Mkdir(req *fuse.MkdirRequest, intr fs.Intr) (fs.Node, fuse.Error) {
	// TODO this lock is too much
	d.fs.mu.Lock()
	defer d.fs.mu.Unlock()

	var snapshot = wiresnap.Snapshot{
		Name: req.Name,
	}
	err := d.fs.db.View(func(tx *bolt.Tx) error {
		return d.rootDir.snapshot(tx, &snapshot.Contents, intr)
	})
	if err != nil {
		return nil, fmt.Errorf("cannot record snapshot: %v", err)
	}

	var key cas.Key
	{
		buf, err := proto.Marshal(&snapshot)
		if err != nil {
			return nil, fmt.Errorf("cannot marshal snapshot: %v", err)
		}
		if len(buf) == 0 {
			return nil, errors.New("marshaled snapshot become empty; this is a bug!")
		}

		// store the snapshot as a chunk, for disaster recovery
		key, err = d.fs.chunkStore.Add(&chunks.Chunk{
			Type:  "snap",
			Level: 0,
			Buf:   buf,
		})
		if err != nil {
			return nil, fmt.Errorf("cannot store snapshot: %v", err)
		}
	}

	var ref = wire.SnapshotRef{
		Key: key,
	}
	buf, err := proto.Marshal(&ref)
	if err != nil {
		return nil, fmt.Errorf("cannot marshal snapshot pointer: %v", err)
	}

	err = d.fs.db.Update(func(tx *bolt.Tx) error {
		b := d.fs.bucket(tx).Bucket(bucketSnap)
		if b == nil {
			return errors.New("snapshot bucket missing")
		}
		return b.Put([]byte(req.Name), buf)
	})

	n, err := snap.Open(d.fs.chunkStore, &snapshot.Contents)
	if err != nil {
		return nil, fmt.Errorf("cannot serve snapshot: %v", err)
	}
	return n, nil
}

// PutProto sets the given key to the protobuf-serialized byte string
// of msg and the provided timestamp.
func PutProto(engine Engine, key Key, msg gogoproto.Message) error {
	data, err := gogoproto.Marshal(msg)
	if err != nil {
		return err
	}
	return engine.Put(key, data)
}

// caller does locking
func (d *dir) saveInternal(tx *bolt.Tx, name string, n node) error {
	if have, ok := d.active[name]; !ok || have != n {
		// unlinked
		return nil
	}

	de, err := n.marshal()
	if err != nil {
		return fmt.Errorf("node save error: %v", err)
	}

	buf, err := proto.Marshal(de)
	if err != nil {
		return fmt.Errorf("Dirent marshal error: %v", err)
	}

	key := pathToKey(d.inode, name)
	bucket := d.fs.bucket(tx).Bucket(bucketDir)
	if bucket == nil {
		return errors.New("dir bucket missing")
	}
	err = bucket.Put(key, buf)
	if err != nil {
		return fmt.Errorf("db write error: %v", err)
	}
	return nil
}

func (e *heartbeatEmitter) generateHeartbeats(heartbeatInterval time.Duration) {
	defer e.instrumentedEmitter.Close()

	ticker := time.NewTicker(heartbeatInterval)
	defer ticker.Stop()
	for {
		select {
		case <-e.stopChan:
			return
		case <-ticker.C:
			hbEvent := e.instrumentedEmitter.GetHeartbeatEvent()
			hbEnvelope, err := Wrap(hbEvent, e.origin)
			if err != nil {
				log.Printf("Failed to wrap heartbeat event: %v\n", err)
				break
			}

			hbData, err := proto.Marshal(hbEnvelope)
			if err != nil {
				log.Printf("Failed to marshal heartbeat event: %v\n", err)
				break
			}

			err = e.innerHbEmitter.Emit(hbData)
			if err != nil {
				log.Printf("Problem while emitting heartbeat data: %v\n", err)
			}
		}
	}
}

func BenchmarkMultiDecodeProtobuf(b *testing.B) {
	b.StopTimer()
	pConfig := NewPipelineConfig(nil) // initializes Globals
	msg := pipeline_ts.GetTestMessage()
	msg.SetPayload("This is a test")
	pack := NewPipelinePack(pConfig.InputRecycleChan())
	pack.MsgBytes, _ = proto.Marshal(msg)
	decoder := new(MultiDecoder)
	decoder.SetPipelineConfig(pConfig)
	conf := decoder.ConfigStruct().(*MultiDecoderConfig)

	RegisterPlugin("ProtobufDecoder", func() interface{} {
		return &ProtobufDecoder{}
	})
	defer delete(AvailablePlugins, "ProtobufDecoder")
	var section PluginConfig
	_, err := toml.Decode("", &section)
	if err != nil {
		b.Fatalf("Error decoding empty TOML: %s", err.Error())
	}
	maker, err := NewPluginMaker("ProtobufDecoder", pConfig, section)
	if err != nil {
		b.Fatalf("Error decoding empty TOML: %s", err.Error())
	}
	pConfig.DecoderMakers["ProtobufDecoder"] = maker

	conf.CascadeStrategy = "first-wins"
	conf.Subs = []string{"sub"}
	decoder.Init(conf)

	b.StartTimer()
	for i := 0; i < b.N; i++ {
		decoder.Decode(pack)
	}
}

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 ProtobufDecoderSpec(c gospec.Context) {
	t := &ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	msg := ts.GetTestMessage()
	config := NewPipelineConfig(nil) // Initializes globals.

	c.Specify("A ProtobufDecoder", func() {
		encoded, err := proto.Marshal(msg)
		c.Assume(err, gs.IsNil)
		pack := NewPipelinePack(config.inputRecycleChan)
		decoder := new(ProtobufDecoder)
		decoder.sampleDenominator = 1000 // Since we don't call decoder.Init().

		c.Specify("decodes a protobuf message", func() {
			pack.MsgBytes = encoded
			_, err := decoder.Decode(pack)
			c.Expect(err, gs.IsNil)
			c.Expect(pack.Message, gs.Equals, msg)
			v, ok := pack.Message.GetFieldValue("foo")
			c.Expect(ok, gs.IsTrue)
			c.Expect(v, gs.Equals, "bar")
		})

		c.Specify("returns an error for bunk encoding", func() {
			bunk := append([]byte{0, 0, 0}, encoded...)
			pack.MsgBytes = bunk
			_, err := decoder.Decode(pack)
			c.Expect(err, gs.Not(gs.IsNil))
		})
	})
}

func BenchmarkMultiDecodeProtobuf(b *testing.B) {
	b.StopTimer()
	pConfig := NewPipelineConfig(nil) // initializes Globals
	msg := pipeline_ts.GetTestMessage()
	msg.SetPayload("This is a test")
	pack := NewPipelinePack(pConfig.InputRecycleChan())
	pack.MsgBytes, _ = proto.Marshal(msg)
	decoder := new(MultiDecoder)
	decoder.SetPipelineConfig(pConfig)
	conf := decoder.ConfigStruct().(*MultiDecoderConfig)
	sub := new(ProtobufDecoder)
	sub.SetPipelineConfig(pConfig)
	sub.Init(nil)
	wrapper0 := NewPluginWrapper("sub", pConfig)
	wrapper0.CreateWithError = func() (interface{}, error) {
		return sub, nil
	}
	pConfig.DecoderWrappers["sub"] = wrapper0
	conf.CascadeStrategy = "first-wins"
	conf.Subs = []string{"sub"}
	decoder.Init(conf)

	b.StartTimer()
	for i := 0; i < b.N; i++ {
		decoder.Decode(pack)
	}
}

func (this *uint32Writer) WriteMsg(msg proto.Message) (err error) {
	var data []byte
	if m, ok := msg.(marshaler); ok {
		n := m.Size()
		if n >= len(this.buffer) {
			this.buffer = make([]byte, n)
		}
		_, err = m.MarshalTo(this.buffer)
		if err != nil {
			return err
		}
		data = this.buffer[:n]
	} else {
		data, err = proto.Marshal(msg)
		if err != nil {
			return err
		}
	}
	length := uint32(len(data))
	if err := binary.Write(this.w, this.byteOrder, &length); err != nil {
		return err
	}
	_, err = this.w.Write(data)
	return err
}

func (this *varintWriter) WriteMsg(msg proto.Message) (err error) {
	var data []byte
	if m, ok := msg.(marshaler); ok {
		n := m.Size()
		if n >= len(this.buffer) {
			this.buffer = make([]byte, n)
		}
		_, err = m.MarshalTo(this.buffer)
		if err != nil {
			return err
		}
		data = this.buffer[:n]
	} else {
		data, err = proto.Marshal(msg)
		if err != nil {
			return err
		}
	}
	length := uint64(len(data))
	n := binary.PutUvarint(this.lenBuf, length)
	_, err = this.w.Write(this.lenBuf[:n])
	if err != nil {
		return err
	}
	_, err = this.w.Write(data)
	return 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
}

// 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))
	}
}

// Encodes the SnapshotRecoveryRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRecoveryRequest) Encode(w io.Writer) (int, error) {

	protoPeers := make([]*protobuf.SnapshotRecoveryRequest_Peer, len(req.Peers))

	for i, peer := range req.Peers {
		protoPeers[i] = &protobuf.SnapshotRecoveryRequest_Peer{
			Name:             proto.String(peer.Name),
			ConnectionString: proto.String(peer.ConnectionString),
		}
	}

	pb := &protobuf.SnapshotRecoveryRequest{
		LeaderName: proto.String(req.LeaderName),
		LastIndex:  proto.Uint64(req.LastIndex),
		LastTerm:   proto.Uint64(req.LastTerm),
		Peers:      protoPeers,
		State:      req.State,
	}
	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

func (proxy *Proxy) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
	r.ParseForm()
	clientAddress := r.RemoteAddr
	appId := r.Form.Get("app")

	authToken := r.Header.Get("Authorization")
	if authToken == "" {
		authToken = extractAuthTokenFromUrl(r.URL)
	}

	ws := upgrade(rw, r)
	authorized, errorMessage := proxy.isAuthorized(appId, authToken, clientAddress)
	if !authorized {
		data, err := proto.Marshal(errorMessage)
		if err != nil {
			proxy.logger.Errorf("Error marshalling log message: %s", err)
		}
		ws.WriteMessage(websocket.BinaryMessage, data)

		ws.Close()
		return
	}

	proxyHandler := NewProxyHandlerProvider(ws, proxy.logger)

	proxyHandler.HandleWebSocket(appId, r.URL.RequestURI(), proxy.hashers)
}

func (me *MsgEncoder) EncodePb(m *PbMessage) error {
	err := me.bw.WriteByte(byte(m.msgType))
	if err != nil {
		return err
	}

	var bytes []byte
	if m.pb != nil {
		bytes, err = proto.Marshal(m.pb)
		if err != nil {
			return err
		}
	}
	size := len(bytes)

	err = binary.Write(me.bw, binary.LittleEndian, uint32(size))

	if err != nil {
		return err
	}

	_, err = me.bw.Write(bytes)

	return me.bw.Flush()
}

func (fh *FakeHandler) handle(conn *websocket.Conn) {
	fh.call()
	request := conn.Request()
	fh.setLastURL(request.URL.String())
	fh.setAuthHeader(request.Header.Get("Authorization"))

	if fh.messageReceived != nil {
		go func() {
			for {
				buffer := make([]byte, 1024)
				_, err := conn.Read(buffer)

				if err == nil {
					fh.messageReceived <- true
				} else {
					break
				}
			}
		}()
	}

	for _, protoMessage := range fh.Messages {
		if protoMessage == nil {
			conn.Write([]byte{})
		} else {
			message, err := proto.Marshal(protoMessage)
			Expect(err).ToNot(HaveOccurred())

			conn.Write(message)
		}
	}

	<-fh.closeConnection
	conn.Close()
}

func (proxy *Proxy) HandleWebSocket(clientWS *websocket.Conn) {
	req := clientWS.Request()
	req.ParseForm()
	req.Form.Get("app")
	clientAddress := clientWS.RemoteAddr()

	appId := req.Form.Get("app")

	extractAuthTokenFromUrl := func(u *url.URL) string {
		authorization := ""
		queryValues := u.Query()
		if len(queryValues["authorization"]) == 1 {
			authorization = queryValues["authorization"][0]
		}
		return authorization
	}

	authToken := clientWS.Request().Header.Get("Authorization")
	if authToken == "" {
		authToken = extractAuthTokenFromUrl(req.URL)
	}

	if authorized, errorMessage := proxy.isAuthorized(appId, authToken, clientAddress); !authorized {
		data, err := proto.Marshal(errorMessage)
		if err != nil {
			proxy.logger.Errorf("Error marshalling log message: %s", err)
		}
		websocket.Message.Send(clientWS, data)
		clientWS.Close()
		return
	}

	defer clientWS.Close()

	proxy.logger.Debugf("Output Proxy: Request for app: %v", req.Form.Get("app"))
	serverWSs := make([]*websocket.Conn, len(proxy.hashers))
	for index, hasher := range proxy.hashers {
		proxy.logger.Debugf("Output Proxy: Servers in group [%v]: %v", index, hasher.LoggregatorServers())

		server := hasher.GetLoggregatorServerForAppId(appId)
		proxy.logger.Debugf("Output Proxy: AppId is %v. Using server: %v", appId, server)

		config, err := websocket.NewConfig("ws://"+server+req.URL.RequestURI(), "http://localhost")

		if err != nil {
			proxy.logger.Errorf("Output Proxy: Error creating config for websocket - %v", err)
		}

		serverWS, err := websocket.DialConfig(config)
		if err != nil {
			proxy.logger.Errorf("Output Proxy: Error connecting to loggregator server - %v", err)
		}

		if serverWS != nil {
			serverWSs[index] = serverWS
		}
	}
	proxy.forwardIO(serverWSs, clientWS)

}

func TestExtend(t *testing.T) {
	fp, err := fieldpath.NewFloat64Path("test", "MyExtendable", test.ThetestDescription(), "FieldA")
	if err != nil {
		panic(err)
	}
	m := &test.MyExtendable{}
	err = proto.SetExtension(m, test.E_FieldA, proto.Float64(10.0))
	if err != nil {
		panic(err)
	}
	buf, err := proto.Marshal(m)
	if err != nil {
		panic(err)
	}
	var unmarshalled float64
	f := FuncHandler{
		Float64Func: func(v float64) {
			t.Logf("unmarshalled %v", v)
			unmarshalled = v
		},
	}
	unmarshaler := fieldpath.NewFloat64Unmarshaler(fp, f)
	err = unmarshaler.Unmarshal(buf)
	if err != nil {
		panic(err)
	}
	if unmarshalled != float64(10.0) {
		panic(fmt.Errorf("wtf %v", unmarshalled))
	}
}