func TestMetadataSerialization(t *testing.T) {
	Convey("Given a metadata structure", t, func() {
		metadata := &Metadata{Name: "nom"}
		marshalled := &bytes.Buffer{}
		metadata.Marshal(marshalled)
		marshalled = bytes.NewBuffer(marshalled.Bytes())

		Convey("The marshalled form should be valid cbor", func() {
			dec := codec.NewDecoder(marshalled, new(codec.CborHandle))
			reheated := &Metadata{}
			dec.MustDecode(reheated)
		})

		Convey("The marshalled form should have known keys", func() {
			dec := codec.NewDecoder(marshalled, new(codec.CborHandle))
			reheated := make(map[string]interface{})
			err := dec.Decode(reheated)
			So(err, ShouldBeNil)
			v, exists := reheated["n"]
			So(exists, ShouldBeTrue)
			So(v, ShouldEqual, "nom")
			So(marshalled.Len(), ShouldEqual, 0)
		})
	})
}

func (c *Client) Stats() (*ClientStats, error) {
	if c.conn == nil {
		return nil, nil
	}

	sess, err := c.Session()
	if err != nil {
		return nil, err
	}

	s, err := sess.Open()
	if err != nil {
		return nil, err
	}

	_, err = s.Write([]byte{uint8(StatsType)})

	buf := []byte{0}

	_, err = io.ReadFull(s, buf)
	if err != nil {
		return nil, err
	}

	switch MessageType(buf[0]) {
	case StatsResultType:
		dec := codec.NewDecoder(s, &msgpack)

		var res ClientStats

		if err := dec.Decode(&res); err != nil {
			return nil, c.checkError(err)
		}

		return &res, nil
	case ErrorType:
		var msgerr Error

		err = codec.NewDecoder(s, &msgpack).Decode(&msgerr)
		if err != nil {
			return nil, c.checkError(err)
		}

		return nil, errors.New(msgerr.Error)
	case SuccessType:
		return nil, nil
	default:
		return nil, c.checkError(EProtocolError)
	}
}

func handleConnection(conn net.Conn, jobd *jobserver.JobServer, cbor *codec.CborHandle) {
	defer conn.Close()

	jobdv := reflect.ValueOf(jobd)

	reader := bufio.NewReader(conn)
	decoder := codec.NewDecoder(reader, cbor)
	writer := bufio.NewWriter(conn)
	encoder := codec.NewEncoder(writer, cbor)

	for {
		var request cborrpc.Request
		err := decoder.Decode(&request)
		if err == io.EOF {
			return
		} else if err != nil {
			fmt.Printf("Error reading message: %v\n", err)
			return
		}
		fmt.Printf("Request: %v\n", request)
		response := doRequest(jobdv, request)
		fmt.Printf("Response: %v\n", response)
		err = encoder.Encode(response)
		if err != nil {
			fmt.Printf("Error encoding response: %v\n", err)
			return
		}
		err = writer.Flush()
		if err != nil {
			fmt.Printf("Error writing response: %v\n", err)
			return
		}
	}
}

func loadPassiveAggressiveStateFormatV1(ctx *core.Context, r io.Reader) (core.SharedState, error) {
	var header regressionMsgpack
	dec := codec.NewDecoder(r, regressionMsgpackHandle)
	if err := dec.Decode(&header); err != nil {
		return nil, err
	}
	if header.Algorithm != "passive_aggressive" {
		return nil, fmt.Errorf("unsupported regression algorithm: %v", header.Algorithm)
	}

	s := &PassiveAggressiveState{}

	var d paStateMsgpack
	if err := dec.Decode(&d); err != nil {
		return nil, err
	}
	s.valueField = d.ValueField
	s.featureVectorField = d.FeatureVectorField

	pa, err := LoadPassiveAggressive(r)
	if err != nil {
		return nil, err
	}
	s.pa = pa
	return s, nil
}

func TestLatestMapEncodingNil(t *testing.T) {
	want := LatestMap{}

	{
		gobs, err := want.GobEncode()
		if err != nil {
			t.Fatal(err)
		}
		have := EmptyLatestMap
		have.GobDecode(gobs)
		if have.Map == nil {
			t.Error("Decoded LatestMap.psMap should not be nil")
		}
	}

	{

		for _, h := range []codec.Handle{
			codec.Handle(&codec.MsgpackHandle{}),
			codec.Handle(&codec.JsonHandle{}),
		} {
			buf := &bytes.Buffer{}
			encoder := codec.NewEncoder(buf, h)
			want.CodecEncodeSelf(encoder)
			decoder := codec.NewDecoder(buf, h)
			have := EmptyLatestMap
			have.CodecDecodeSelf(decoder)
			if !reflect.DeepEqual(want, have) {
				t.Error(test.Diff(want, have))
			}
		}
	}
}

func TestSpanMsgPack(t *testing.T) {
	span := Span{Id: TestId("33f25a1a750a471db5bafa59309d7d6f"),
		SpanData: SpanData{
			Begin:       1234,
			End:         5678,
			Description: "getFileDescriptors",
			Parents:     []SpanId{},
			TracerId:    "testTracerId",
		}}
	mh := new(codec.MsgpackHandle)
	mh.WriteExt = true
	w := bytes.NewBuffer(make([]byte, 0, 2048))
	enc := codec.NewEncoder(w, mh)
	err := enc.Encode(span)
	if err != nil {
		t.Fatal("Error encoding span as msgpack: " + err.Error())
	}
	buf := w.Bytes()
	fmt.Printf("span: %s\n", hex.EncodeToString(buf))
	mh = new(codec.MsgpackHandle)
	mh.WriteExt = true
	dec := codec.NewDecoder(bytes.NewReader(buf), mh)
	var span2 Span
	err = dec.Decode(&span2)
	if err != nil {
		t.Fatal("Failed to reverse msgpack encoding for " + span.String())
	}
	ExpectSpansEqual(t, &span, &span2)
}

func TestEdgeMetadatasEncodingNil(t *testing.T) {
	want := EdgeMetadatas{}

	{
		gobs, err := want.GobEncode()
		if err != nil {
			t.Fatal(err)
		}
		have := EmptyEdgeMetadatas
		have.GobDecode(gobs)
		if have.psMap == nil {
			t.Error("needed to get back a non-nil psMap for EdgeMetadata")
		}
	}

	{

		for _, h := range []codec.Handle{
			codec.Handle(&codec.MsgpackHandle{}),
			codec.Handle(&codec.JsonHandle{}),
		} {
			buf := &bytes.Buffer{}
			encoder := codec.NewEncoder(buf, h)
			want.CodecEncodeSelf(encoder)
			decoder := codec.NewDecoder(buf, h)
			have := EmptyEdgeMetadatas
			have.CodecDecodeSelf(decoder)
			if !reflect.DeepEqual(want, have) {
				t.Error(test.Diff(want, have))
			}
		}
	}
}

// This test determines the behavior of codec with respect to advancing the
// Reader when decoding error scenarios. It seems that the codec advances
// the Reader if Decode fails, but sets its own state to expect a specific
// type for the next Decode, and thus is functionally the same as not
// advancing the Reader.
func TestCodec(t *testing.T) {
	var buf bytes.Buffer
	mh := &codec.MsgpackHandle{WriteExt: true}
	enc := codec.NewEncoder(&buf, mh)
	dec := codec.NewDecoder(&buf, mh)

	var i int = math.MaxInt32
	err := enc.Encode(i)
	require.Nil(t, err, "expected encoding to succeed")
	require.Equal(t, 5, len(buf.Bytes()), "expected buffer to contain bytes")

	var targetInt int
	err = dec.Decode(&targetInt)
	require.Nil(t, err, "expected decoding to succeed")
	require.Equal(t, math.MaxInt32, targetInt, "expected codec to successfully decode int")
	require.Equal(t, 0, len(buf.Bytes()), "expected buffer to be empty")

	var targetString string
	enc.Encode(i)
	require.Equal(t, 5, len(buf.Bytes()), "expected buffer to contain bytes")
	err = dec.Decode(&targetString)
	require.Error(t, err, "expected error while decoding")
	require.Contains(t, err.Error(), "Unrecognized descriptor byte", "expected error while decoding")
	require.Equal(t, 4, len(buf.Bytes()), "expected buffer to have bytes")
	err = dec.Decode(&targetString)
	require.Error(t, err, "expected error while decoding")
	require.Contains(t, err.Error(), "Unrecognized descriptor byte", "expected error while decoding")
	require.Equal(t, 4, len(buf.Bytes()), "expected buffer to have bytes")

	targetInt = 0
	err = dec.Decode(&targetInt)
	require.Nil(t, err, "expected decoding to succeed")
	require.Equal(t, math.MaxInt32, targetInt, "expected codec to successfully decode int")
	require.Equal(t, 0, len(buf.Bytes()), "expected buffer to be empty")
}

// Decode tries to decode a restdata object from a reader, such as an
// HTTP request or response.  out must be a pointer type.
func Decode(contentType string, r io.Reader, out interface{}) error {
	if contentType == "" {
		// RFC 7231 section 3.1.1.5
		// We could also consider http.DetectContentType()
		contentType = "application/octet-stream"
	}

	mediaType, _, err := mime.ParseMediaType(contentType)
	if err != nil {
		return err
	}

	// Promote to more specific types
	switch mediaType {
	case "text/json", "application/json", JSONMediaType, V1JSONMediaType:
		mediaType = V1JSONMediaType

	default:
		return ErrUnsupportedMediaType{Type: mediaType}
	}

	// Actually decode the object based on the selected type.
	switch mediaType {
	case V1JSONMediaType:
		// (We will be happy we picked this library if we do
		// CBOR over the wire; it is also used for the
		// CBOR-RPC Python compatibility interface)
		json := &codec.JsonHandle{}
		decoder := codec.NewDecoder(r, json)
		err = decoder.Decode(out)
	default:
		err = ErrUnsupportedMediaType{Type: mediaType}
	}
	return err
}

func (c *duplexconn) readFrames() {
	reader := bufio.NewReader(c.conn)
	decoder := codec.NewDecoder(reader, &mh)
	for {
		var frame Frame
		err := decoder.Decode(&frame)
		if err != nil {
			debug(err)
			close(c.writeCh)
			c.peer.Drop(c.uuid)
			return
		}
		debug("Reading frame:", frame)
		c.Lock()
		ch, exists := c.channels[frame.Channel]
		c.Unlock()
		if exists && frame.Type == DataFrame {
			if ch.HandleIncoming(&frame) {
				continue
			}
		}
		if !exists && frame.Type == OpenFrame {
			if c.peer.HandleOpen(c, &frame) {
				continue
			}
		}
		debug("Dropped frame:", frame)
	}
}

// getConn is used to get a connection from the pool
func (n *NetworkTransport) getConn(target net.Addr) (*netConn, error) {
	// Check for a pooled conn
	if conn := n.getPooledConn(target); conn != nil {
		return conn, nil
	}

	// Dial a new connection
	conn, err := n.stream.Dial(target.String(), n.timeout)
	if err != nil {
		return nil, err
	}

	// Wrap the conn
	netConn := &netConn{
		target: target,
		conn:   conn,
		r:      bufio.NewReader(conn),
		w:      bufio.NewWriter(conn),
	}

	// Setup encoder/decoders
	netConn.dec = codec.NewDecoder(netConn.r, &codec.MsgpackHandle{})
	netConn.enc = codec.NewEncoder(netConn.w, &codec.MsgpackHandle{})

	// Done
	return netConn, nil
}

func TestHTTPPollMailboxMsgPack(t *testing.T) {
	reg := NewMemRegistry()
	serv := NewHTTPService(cPort, reg)

	reg.Declare("a")

	msg := Msg("hello")

	reg.Push("a", msg)

	url := fmt.Sprintf("http://%s/mailbox/a", cPort)

	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		panic(err)
	}

	req.Header.Add("Accept", ctMsgPack)

	rw := httptest.NewRecorder()

	serv.mux.ServeHTTP(rw, req)

	assert.Equal(t, 200, rw.Code, "server error")

	var ret Message

	err = codec.NewDecoder(rw.Body, &msgpack).Decode(&ret)
	if err != nil {
		panic(err)
	}

	assert.True(t, msg.Equal(&ret), "poll did not return a message")
}

func ReadMessage(conn net.Conn, timeout time.Duration) (*Message, error) {
	conn.SetReadDeadline(time.Now().Add(DefaultHeartbeatTimeout))
	var lengthBuf [2]byte
	_, err := conn.Read(lengthBuf[:])
	if err != nil {
		return nil, err
	}

	length := int(binary.BigEndian.Uint16(lengthBuf[:]))
	if length > DefaultMessageLengthCap {
		return nil, errors.New("message is too long")
	}

	buf := bytes.NewBuffer(make([]byte, 0, length))
	_, err = io.CopyN(buf, conn, int64(length))
	if err != nil {
		return nil, err
	}

	var handle = &codec.MsgpackHandle{}
	var decoder = codec.NewDecoder(buf, handle)
	var result = &Message{}
	err = decoder.Decode(result)
	if err != nil {
		return nil, err
	}

	return result, nil
}

func loadAROWStateFormatV1(ctx *core.Context, r io.Reader) (core.SharedState, error) {
	var header classifierMsgpack
	dec := codec.NewDecoder(r, classifierMsgpackHandle)
	if err := dec.Decode(&header); err != nil {
		return nil, err
	}
	if header.Algorithm != "arow" {
		return nil, fmt.Errorf("unsupported classification algorithm: %v", header.Algorithm)
	}

	// This is the current format and no data type conversion is required.
	s := &AROWState{}

	var d arowStateMsgpack
	if err := dec.Decode(&d); err != nil {
		return nil, err
	}
	s.labelField = d.LabelField
	s.featureVectorField = d.FeatureVectorField

	arow, err := LoadAROW(r)
	if err != nil {
		return nil, err
	}
	s.arow = arow
	return s, nil
}

func (r *Request) Decode(t interface{}) {
	buf := bytes.NewBuffer(r.params)
	dec := codec.NewDecoder(buf, &mh)
	dec.Decode(t)

	return
}

func main() {

	var m = MsgPackSample{
		Name:    "ユイにゃん",
		Num:     17,
		Message: "AngelBeats!",
	}

	mh := &codec.MsgpackHandle{RawToString: true}
	buf1 := &bytes.Buffer{}
	buf2 := &bytes.Buffer{}

	var msgpackExample = MsgPackExample{
		mh:   mh,
		buf1: buf1,
		buf2: buf2,
		enc:  codec.NewEncoder(buf1, mh),
		dec:  codec.NewDecoder(buf2, mh),
		menc: msgpack.NewEncoder(buf1),
		mdec: msgpack.NewDecoder(buf2),
	}

	msgpackExample.packUnPackMsgPack(m)

	msgpackExample.packUnPackCodec(m)
}

func (i *ForwardInput) handleConnection(rw io.ReadWriter) {
	dec := codec.NewDecoder(rw, mh)
	for {
		var v []interface{}
		err := dec.Decode(&v)
		if err != nil {
			if err != io.EOF {
				i.env.Log.Warning(err)
			}
			return
		}

		tag := v[0].(string)
		switch len(v) {
		case 2:
			i.parseNestedEncoding(tag, v)
		case 3:
			// 1 is timeBinaryVersion defined in time.go.
			// In msgpack spec, 0x01 means positive fixint, thus
			// it's never appeared in this context.
			if s, ok := v[1].(string); ok && s[0] != 1 {
				i.parseNestedEncoding(tag, v)
				i.handleOption(rw, v[2].(map[string]interface{}))
			} else {
				i.parseFlatEncoding(tag, v)
			}
		case 4:
			i.parseFlatEncoding(tag, v)
			i.handleOption(rw, v[3].(map[string]interface{}))
		}
	}
}

// NewRPCClient is used to create a new RPC client given the address.
// This will properly dial, handshake, and start listening
func NewRPCClient(addr string) (*RPCClient, error) {
	// Try to dial to serf
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		return nil, err
	}

	// Create the client
	client := &RPCClient{
		seq:        0,
		conn:       conn.(*net.TCPConn),
		reader:     bufio.NewReader(conn),
		writer:     bufio.NewWriter(conn),
		dispatch:   make(map[uint64]seqHandler),
		shutdownCh: make(chan struct{}),
	}
	client.dec = codec.NewDecoder(client.reader,
		&codec.MsgpackHandle{RawToString: true, WriteExt: true})
	client.enc = codec.NewEncoder(client.writer,
		&codec.MsgpackHandle{RawToString: true, WriteExt: true})
	go client.listen()

	// Do the initial handshake
	if err := client.handshake(); err != nil {
		client.Close()
		return nil, err
	}
	return client, err
}

// ReadBinary reads bytes into a Report.
//
// Will decompress the binary if gzipped is true, and will use the given
// codecHandle to decode it.
func (rep *Report) ReadBinary(r io.Reader, gzipped bool, codecHandle codec.Handle) error {
	var err error
	var compressedSize, uncompressedSize uint64

	// We have historically had trouble with reports being too large. We are
	// keeping this instrumentation around to help us implement
	// weaveworks/scope#985.
	if log.GetLevel() == log.DebugLevel {
		r = byteCounter{next: r, count: &compressedSize}
	}
	if gzipped {
		r, err = gzip.NewReader(r)
		if err != nil {
			return err
		}
	}
	if log.GetLevel() == log.DebugLevel {
		r = byteCounter{next: r, count: &uncompressedSize}
	}
	if err := codec.NewDecoder(r, codecHandle).Decode(&rep); err != nil {
		return err
	}
	log.Debugf(
		"Received report sizes: compressed %d bytes, uncompressed %d bytes (%.2f%%)",
		compressedSize,
		uncompressedSize,
		float32(compressedSize)/float32(uncompressedSize)*100,
	)
	return nil
}

func handleReq(params *DummyServerParams, resp http.ResponseWriter, req *http.Request, matchparams map[string]string) {
	resp.Header().Set("Content-Type", "application/json; charset=UTF-8")
	h := md5.New()
	format := matchparams["format"]
	var body []byte
	var err error
	if params.ReadThrottle > 0 {
		body, err = ReadThrottled(req.Body, int(req.ContentLength), params.ReadThrottle)
	} else {
		body, err = ioutil.ReadAll(req.Body)
	}
	if err != nil {
		internalServerError(resp)
		return
	}
	rdr := (io.Reader)(bytes.NewReader(body))
	if format == "msgpack.gz" {
		rdr, err = gzip.NewReader(rdr)
		if err != nil {
			internalServerError(resp)
			return
		}
	}
	_codec := &codec.MsgpackHandle{}
	decoder := codec.NewDecoder(rdr, _codec)
	numRecords := 0
	for {
		v := map[string]interface{}{}
		err := decoder.Decode(&v)
		if err != nil {
			if err == io.EOF {
				break
			} else {
				Error("%s", err.Error())
				break
			}
		}
		numRecords += 1
	}
	fmt.Printf("%d records received\n", numRecords)
	io.Copy(h, bytes.NewReader(body))
	md5sum := make([]byte, 0, h.Size())
	md5sum = h.Sum(md5sum)
	uniqueId, _ := matchparams["uniqueId"]
	respData := map[string]interface{}{
		"unique_id":    uniqueId,
		"database":     matchparams["database"],
		"table":        matchparams["table"],
		"md5_hex":      hex.EncodeToString(md5sum),
		"elapsed_time": 0.,
	}
	payload, err := json.Marshal(respData)
	if err != nil {
		internalServerError(resp)
		return
	}
	resp.WriteHeader(200)
	resp.Write(payload)
}