func (c *EncodedConn) FlushTimeout(timeout time.Duration) (err error) {
	return c.Conn.FlushTimeout(timeout)
}
FlushTimeout allows a Flush operation to have an associated timeout.
package nats
import (
	"errors"
	"fmt"
	"reflect"
	"sync"
	"time"
)
type Encoder interface {
	Encode(subject string, v interface{}) ([]byte, error)
	Decode(subject string, data []byte, vPtr interface{}) error
}
var encMap map[string]Encoder
var encLock sync.Mutex
const (
	JSON_ENCODER    = "json"
	GOB_ENCODER     = "gob"
	DEFAULT_ENCODER = "default"
)
func init() {
	encMap = make(map[string]Encoder)
	RegisterEncoder("json", &JsonEncoder{})
	RegisterEncoder("gob", &GobEncoder{})
	RegisterEncoder("default", &DefaultEncoder{})
}
type EncodedConn struct {
	Conn *Conn
	Enc  Encoder
}
func NewEncodedConn(c *Conn, encType string) (*EncodedConn, error) {
	if c == nil {
		return nil, errors.New("nats: Nil Connection")
	}
	if c.IsClosed() {
		return nil, ErrConnectionClosed
	}
	ec := &EncodedConn{Conn: c, Enc: EncoderForType(encType)}
	if ec.Enc == nil {
		return nil, fmt.Errorf("No encoder registered for '%s'", encType)
	}
	return ec, nil
}
func RegisterEncoder(encType string, enc Encoder) {
	encLock.Lock()
	defer encLock.Unlock()
	encMap[encType] = enc
}
func EncoderForType(encType string) Encoder {
	encLock.Lock()
	defer encLock.Unlock()
	return encMap[encType]
}
func (c *EncodedConn) Publish(subject string, v interface{}) error {
	b, err := c.Enc.Encode(subject, v)
	if err != nil {
		c.Conn.err = err
		return err
	}
	return c.Conn.publish(subject, _EMPTY_, b)
}
func (c *EncodedConn) PublishRequest(subject, reply string, v interface{}) error {
	b, err := c.Enc.Encode(subject, v)
	if err != nil {
		return err
	}
	return c.Conn.publish(subject, reply, b)
}
func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, timeout time.Duration) error {
	b, err := c.Enc.Encode(subject, v)
	if err != nil {
		return err
	}
	m, err := c.Conn.Request(subject, b, timeout)
	if err != nil {
		return err
	}
	if reflect.TypeOf(vPtr) == emptyMsgType {
		mPtr := vPtr.(*Msg)
		*mPtr = *m
	} else {
		err = c.Enc.Decode(m.Subject, m.Data, vPtr)
	}
	return err
}
type Handler interface{}
func argInfo(cb Handler) (reflect.Type, int) {
	cbType := reflect.TypeOf(cb)
	if cbType.Kind() != reflect.Func {
		panic("nats: Handler needs to be a func")
	}
	numArgs := cbType.NumIn()
	if numArgs == 0 {
		return nil, numArgs
	}
	return cbType.In(numArgs - 1), numArgs
}
var emptyMsgType = reflect.TypeOf(&Msg{})
func (c *EncodedConn) Subscribe(subject string, cb Handler) (*Subscription, error) {
	return c.subscribe(subject, _EMPTY_, cb)
}
func (c *EncodedConn) QueueSubscribe(subject, queue string, cb Handler) (*Subscription, error) {
	return c.subscribe(subject, queue, cb)
}
func (c *EncodedConn) subscribe(subject, queue string, cb Handler) (*Subscription, error) {
	argType, numArgs := argInfo(cb)
	cbValue := reflect.ValueOf(cb)
	wantsRaw := (argType == emptyMsgType)
	natsCB := func(m *Msg) {
		var oV []reflect.Value
		if wantsRaw {
			oV = []reflect.Value{reflect.ValueOf(m)}
		} else {
			var oPtr reflect.Value
			if argType.Kind() != reflect.Ptr {
				oPtr = reflect.New(argType)
			} else {
				oPtr = reflect.New(argType.Elem())
			}
			if err := c.Enc.Decode(m.Subject, m.Data, oPtr.Interface()); err != nil {
				nc := c.Conn
				nc.mu.Lock()
				nc.err = errors.New("nats: Got an error trying to unmarshal: " + err.Error())
				if nc.Opts.AsyncErrorCB != nil {
					go c.Conn.Opts.AsyncErrorCB(c.Conn, m.Sub, c.Conn.err)
				}
				nc.mu.Unlock()
				return
			}
			if argType.Kind() != reflect.Ptr {
				oPtr = reflect.Indirect(oPtr)
			}
			switch numArgs {
			case 1:
				oV = []reflect.Value{oPtr}
			case 2:
				subV := reflect.ValueOf(m.Subject)
				oV = []reflect.Value{subV, oPtr}
			case 3:
				subV := reflect.ValueOf(m.Subject)
				replyV := reflect.ValueOf(m.Reply)
				oV = []reflect.Value{subV, replyV, oPtr}
			}
		}
		cbValue.Call(oV)
	}
	return c.Conn.subscribe(subject, queue, natsCB, c.Conn.Opts.SubChanLen)
}