// Shows different ways to create a Conn
func ExampleConnect() {

	nats.Connect(nats.DefaultURL)
	nats.Connect("nats://derek:[email protected]:421")

	opts := nats.Options{
		AllowReconnect: true,
		MaxReconnect:   10,
		ReconnectWait:  5 * time.Second,
		Timeout:        1 * time.Second,
	}

	nc, _ := opts.Connect()
	nc.Close()
}

func NewGnatsd(numberOfMessages int, testLatency bool) *Gnatsd {
	conn, _ := nats.Connect(nats.DefaultURL)

	// We want to be alerted if we get disconnected, this will
	// be due to Slow Consumer.
	conn.Opts.AllowReconnect = false

	// Report async errors.
	conn.Opts.AsyncErrorCB = func(nc *nats.Conn, sub *nats.Subscription, err error) {
		panic(fmt.Sprintf("NATS: Received an async error! %v\n", err))
	}

	// Report a disconnect scenario.
	conn.Opts.DisconnectedCB = func(nc *nats.Conn) {
		fmt.Printf("Getting behind! %d\n", nc.OutMsgs-nc.InMsgs)
		panic("NATS: Got disconnected!")
	}

	var handler benchmark.MessageHandler
	if testLatency {
		handler = &benchmark.LatencyMessageHandler{
			NumberOfMessages: numberOfMessages,
			Latencies:        []float32{},
		}
	} else {
		handler = &benchmark.ThroughputMessageHandler{NumberOfMessages: numberOfMessages}
	}

	return &Gnatsd{
		handler:     handler,
		subject:     "test",
		conn:        conn,
		testLatency: testLatency,
	}
}

func ExampleConn_PublishMsg() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	msg := &nats.Msg{Subject: "foo", Reply: "bar", Data: []byte("Hello World!")}
	nc.PublishMsg(msg)
}

func (n *ntport) Dial(addr string, opts ...transport.DialOption) (transport.Client, error) {
	cAddr := nats.DefaultURL

	if len(n.addrs) > 0 && strings.HasPrefix(n.addrs[0], "nats://") {
		cAddr = n.addrs[0]
	}

	c, err := nats.Connect(cAddr)
	if err != nil {
		return nil, err
	}

	id := nats.NewInbox()
	sub, err := c.SubscribeSync(id)
	if err != nil {
		return nil, err
	}

	return &ntportClient{
		conn: c,
		addr: addr,
		id:   id,
		sub:  sub,
	}, nil
}

// This Example shows an asynchronous subscriber.
func ExampleConn_Subscribe() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	nc.Subscribe("foo", func(m *nats.Msg) {
		fmt.Printf("Received a message: %s\n", string(m.Data))
	})
}

func ExampleSubscription_Unsubscribe() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	sub, _ := nc.SubscribeSync("foo")
	// ...
	sub.Unsubscribe()
}

func ExampleConn_Request() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	nc.Subscribe("foo", func(m *nats.Msg) {
		nc.Publish(m.Reply, []byte("I will help you"))
	})
	nc.Request("foo", []byte("help"), 50*time.Millisecond)
}

func main() {
	nc, _ := nats.Connect("nats://yourhost:4222")
	defer nc.Close()
	nc.QueueSubscribe(TOPIC, QUEUE, func(m *nats.Msg) {
		fmt.Println(string(m.Data))
		open.Start(string(m.Data))
	})
	select {}
}

func (b Bopen) Open(url string) {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()
	if b.Topic != "" {
		nc.Publish(b.Topic, []byte(url))
	} else {
		nc.Publish("url", []byte(url))
	}
}

func ExampleConn_QueueSubscribe() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	received := 0

	nc.QueueSubscribe("foo", "worker_group", func(_ *nats.Msg) {
		received += 1
	})
}

func (nc *NatsClient) Init(server, service string) {
	var err error
	nc.natsConn, err = nats.Connect(server)
	if err != nil {
		panic(err)
	}
	nc.natsEnConn, err = nats.NewEncodedConn(nc.natsConn, nats.JSON_ENCODER)
	if err != nil {
		panic(err)
	}
	//defer ec.Close()
}

func ExampleSubscription_NextMsg() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	sub, _ := nc.SubscribeSync("foo")
	m, err := sub.NextMsg(1 * time.Second)
	if err == nil {
		fmt.Printf("Received a message: %s\n", string(m.Data))
	} else {
		fmt.Println("NextMsg timed out.")
	}
}

func ExampleConn_Flush() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	msg := &nats.Msg{Subject: "foo", Reply: "bar", Data: []byte("Hello World!")}
	for i := 0; i < 1000; i++ {
		nc.PublishMsg(msg)
	}
	err := nc.Flush()
	if err == nil {
		// Everything has been processed by the server for nc *Conn.
	}
}

// EncodedConn can publish virtually anything just
// by passing it in. The encoder will be used to properly
// encode the raw Go type
func ExampleEncodedConn_Publish() {
	nc, _ := nats.Connect(nats.DefaultURL)
	c, _ := nats.NewEncodedConn(nc, "json")
	defer c.Close()

	type person struct {
		Name    string
		Address string
		Age     int
	}

	me := &person{Name: "derek", Age: 22, Address: "85 Second St"}
	c.Publish("hello", me)
}

// BindSendChan() allows binding of a Go channel to a nats
// subject for publish operations. The Encoder attached to the
// EncodedConn will be used for marshalling.
func ExampleEncodedConn_BindSendChan() {
	nc, _ := nats.Connect(nats.DefaultURL)
	c, _ := nats.NewEncodedConn(nc, "json")
	defer c.Close()

	type person struct {
		Name    string
		Address string
		Age     int
	}

	ch := make(chan *person)
	c.BindSendChan("hello", ch)

	me := &person{Name: "derek", Age: 22, Address: "85 Second St"}
	ch <- me
}

func (n *ntport) Listen(addr string) (transport.Listener, error) {
	cAddr := nats.DefaultURL

	if len(n.addrs) > 0 && strings.HasPrefix(n.addrs[0], "nats://") {
		cAddr = n.addrs[0]
	}

	c, err := nats.Connect(cAddr)
	if err != nil {
		return nil, err
	}

	return &ntportListener{
		addr: nats.NewInbox(),
		conn: c,
		exit: make(chan bool, 1),
	}, nil
}

func ExampleSubscription_AutoUnsubscribe() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	received, wanted, total := 0, 10, 100

	sub, _ := nc.Subscribe("foo", func(_ *nats.Msg) {
		received += 1
	})
	sub.AutoUnsubscribe(wanted)

	for i := 0; i < total; i++ {
		nc.Publish("foo", []byte("Hello"))
	}
	nc.Flush()

	fmt.Printf("Received = %d", received)
}

// NewPeer creates and returns a new Peer for communicating with NATS.
func NewPeer(host string) (*Peer, error) {
	conn, err := nats.Connect(fmt.Sprintf("nats://%s", host))
	if err != nil {
		return nil, err
	}

	// We want to be alerted if we get disconnected, this will be due to Slow
	// Consumer.
	conn.Opts.AllowReconnect = false

	return &Peer{
		conn:     conn,
		messages: make(chan []byte, 10000),
		send:     make(chan []byte),
		errors:   make(chan error, 1),
		done:     make(chan bool),
	}, nil
}

func init() {
	cfgPtr := flag.String("config", "config/director.json", "Path to the config file")
	flag.Parse()

	f, err := os.Open(*cfgPtr)
	if err != nil {
		fmt.Println("Cannot open the config file:", err)
		os.Exit(1)
	}

	err = json.NewDecoder(f).Decode(&cfg)
	if err != nil {
		fmt.Println("Cannot decode the config file:", err)
		os.Exit(1)
	}

	db_dsn := cfg.DB["Username"] + ":" + cfg.DB["Password"] + "@tcp(" + cfg.DB["Host"] + ":" + cfg.DB["Port"] + ")/" + cfg.DB["Database"]
	db, err = sql.Open("mysql", db_dsn)
	if err != nil {
		fmt.Println("Cannot connect to the Policy database:", err)
		os.Exit(1)
	}

	if cfg.Duration != 0 {
		duration = cfg.Duration
	}

	natsc, err = nats.Connect(cfg.Nats)
	if err != nil {
		fmt.Println("Cannot connect to the gnatsd:", err)
		os.Exit(1)
	}

	if cfg.Log != "" {
		logf, err := os.Open(cfg.Log)
		if err != nil {
			fmt.Println("Cannot open the log file:", err)
			os.Exit(1)
		}
		log.SetOutput(logf)
	}
}

// Connect to rabbitmq
func (b *Broker) Connect(uri string) error {
	b.natsURL = uri
	log.Debugf("Dialing [%s]", uri)

	// dial the server
	conn, err := nats.Connect(b.natsURL)
	if err != nil {
		return err
	}

	// create the encoded connection
	b.connection, err = nats.NewEncodedConn(conn, "json")
	if err != nil {
		return err
	}

	log.Debug("Connected to gnatsd")

	return nil
}