func init() {
	recordsRead = expvar.NewInt("RecordsRead")
	recordsWritten = expvar.NewInt("RecordsWritten")
	bytesRead = expvar.NewInt("BytesRead")
	bytesWritten = expvar.NewInt("BytesWritten")
	seeks = expvar.NewInt("Seeks")
}

func init() {
	stats.cacheTotal = expvar.NewInt("cache-total")
	stats.cacheBypassed = expvar.NewInt("cache-bypassed")
	stats.cacheHits = expvar.NewInt("cache-hits")
	stats.cacheMisses = expvar.NewInt("cache-misses")
	stats.cacheRecorded = expvar.NewInt("cache-recorded")
}

func NewPubMetrics(gw *Gateway) *pubMetrics {
	this := &pubMetrics{
		gw:         gw,
		PubOkMap:   make(map[string]metrics.Counter),
		PubFailMap: make(map[string]metrics.Counter),

		ClientError: metrics.NewRegisteredCounter("pub.clienterr", metrics.DefaultRegistry),
		PubQps:      metrics.NewRegisteredMeter("pub.qps", metrics.DefaultRegistry),
		PubTryQps:   metrics.NewRegisteredMeter("pub.try.qps", metrics.DefaultRegistry),
		JobQps:      metrics.NewRegisteredMeter("job.qps", metrics.DefaultRegistry),
		JobTryQps:   metrics.NewRegisteredMeter("job.try.qps", metrics.DefaultRegistry),
		PubMsgSize:  metrics.NewRegisteredHistogram("pub.msgsize", metrics.DefaultRegistry, metrics.NewExpDecaySample(1028, 0.015)),
		JobMsgSize:  metrics.NewRegisteredHistogram("job.msgsize", metrics.DefaultRegistry, metrics.NewExpDecaySample(1028, 0.015)),
		PubLatency:  metrics.NewRegisteredHistogram("pub.latency", metrics.DefaultRegistry, metrics.NewExpDecaySample(1028, 0.015)),
	}

	if Options.DebugHttpAddr != "" {
		this.expPubOk = expvar.NewInt("PubOk")
		this.expPubFail = expvar.NewInt("PubFail")
		this.expActiveConns = expvar.NewInt("PubConns")       // TODO
		this.expActiveUpstream = expvar.NewInt("PubUpstream") // TODO
	}

	return this
}

/*
 * Initializations
 */
func init() {
	flag.Parse()
	status.InputEventCount = expvar.NewInt("input_event_count")
	status.OutputEventCount = expvar.NewInt("output_event_count")
	status.ErrorCount = expvar.NewInt("error_count")

	expvar.Publish("connection_status",
		expvar.Func(func() interface{} {
			res := make(map[string]interface{}, 0)
			res["last_connect_time"] = status.LastConnectTime
			res["last_error_text"] = status.LastConnectError
			res["last_error_time"] = status.ErrorTime
			if status.IsConnected {
				res["connected"] = true
				res["uptime"] = time.Now().Sub(status.LastConnectTime).Seconds()
			} else {
				res["connected"] = false
				res["uptime"] = 0.0
			}

			return res
		}))
	expvar.Publish("uptime", expvar.Func(func() interface{} {
		return time.Now().Sub(status.StartTime).Seconds()
	}))
	expvar.Publish("subscribed_events", expvar.Func(func() interface{} {
		return config.EventTypes
	}))

	results = make(chan string, 100)
	output_errors = make(chan error)

	status.StartTime = time.Now()
}

//NewMetrics creates
//Metrics object
func NewMetrics() *Metrics {
	metrics = &Metrics{
		true,
		expvar.NewInt("comparisons"),
		expvar.NewInt("errors"),
		expvar.NewString("database"),
	}
	return metrics
}

func main() {
	region := "us-east-1"
	queueName := "example_queue"
	numFetchers := 3

	// set up an SQS service instance
	// note that you can modify the AWS config used - make your own sqsconsumer.AWSConfigOption
	// or just depend on ~/.aws/... or environment variables and don't pass any opts at all
	s, err := sqsconsumer.SQSServiceForQueue(queueName, sqsconsumer.OptAWSRegion(region))
	if err != nil {
		log.Fatalf("Could not set up queue '%s': %s", queueName, err)
	}

	// set up a context which will gracefully cancel the worker on interrupt
	fetchCtx, cancelFetch := context.WithCancel(context.Background())
	term := make(chan os.Signal, 1)
	signal.Notify(term, os.Interrupt, os.Kill)
	go func() {
		<-term
		log.Println("Starting graceful shutdown")
		cancelFetch()
	}()

	// set up metrics - note TrackMetrics does not run the http server, and uses expvar
	exposeMetrics()
	ms := expvar.NewInt(fmt.Sprintf("%s.success", queueName))
	mf := expvar.NewInt(fmt.Sprintf("%s.fail", queueName))
	mt := expvar.NewFloat(fmt.Sprintf("%s.time", queueName))
	track := middleware.TrackMetrics(ms, mf, mt)

	// wrap the handler
	handler := middleware.ApplyDecoratorsToHandler(processMessage, track)

	// start the consumers
	log.Println("Starting queue consumers")

	wg := &sync.WaitGroup{}
	wg.Add(numFetchers)
	for i := 0; i < numFetchers; i++ {
		go func() {
			// create the consumer and bind it to a queue and processor function
			c := sqsconsumer.NewConsumer(s, handler)

			// start running the consumer with a context that will be cancelled when a graceful shutdown is requested
			c.Run(fetchCtx)

			wg.Done()
		}()
	}

	// wait for all the consumers to exit cleanly
	wg.Wait()
	log.Println("Shutdown complete")
}

func main() {
	countPckRecv := expvar.NewInt(keyCountPckRecv)
	countPckRecv.Set(0)

	countPckSent := expvar.NewInt(keyCountPckSent)
	countPckSent.Set(0)

	countPckErr := expvar.NewInt(keyCountPckErr)
	countPckErr.Set(0)

	UserMap := NewUsers(0, 100)

	http.HandleFunc(userPrefix, func(w http.ResponseWriter, r *http.Request) {
		countPckRecv.Add(1)

		uid := strings.TrimPrefix(r.URL.Path, userPrefix)

		w.Header().Set("Content-Type", "application/json")

		u := UserMap.Get(uid)
		ret := 0
		switch r.Method {
		case "GET":
			if u == nil {
				ret = 2
			}
			res := Response{ReturnCode: ret, Data: u}

			err := json.NewEncoder(w).Encode(res)
			if err != nil {
				http.Error(w, fmt.Sprintf("%s", err), http.StatusInternalServerError)
				return
			}
			// 		case "PUT":
			// 			if u == nil {
			// 				http.Error(w, fmt.Sprintf("%s not found", uid), http.StatusInternalServerError)
			// 				return
			// 			}

		case "POST":
		case "DELETE":
		default:
			http.Error(w, "Method not allow", http.StatusMethodNotAllowed)
			return
		}

		countPckSent.Add(1)
		return
	})

	log.Printf("Running server on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

func NewHelloWorldModule() *HelloWorldModule {

	var cfg Config

	ok := logging.ReadModuleConfig(&cfg, "config", "hello") || logging.ReadModuleConfig(&cfg, "files/etc/gosample", "hello")
	if !ok {
		// when the app is run with -e switch, this message will automatically be redirected to the log file specified
		log.Fatalln("failed to read config")
	}

	// this message only shows up if app is run with -debug option, so its great for debugging
	logging.Debug.Println("hello init called", cfg.Server.Name)

	// contohnya: caranya ciptakan nsq consumer
	nsqCfg := nsq.NewConfig()
	q := createNewConsumer(nsqCfg, "random-topic", "test", handler)
	q.SetLogger(log.New(os.Stderr, "nsq:", log.Ltime), nsq.LogLevelError)
	q.ConnectToNSQLookupd("nsqlookupd.local:4161")

	return &HelloWorldModule{
		cfg:       &cfg,
		something: "John Doe",
		stats:     expvar.NewInt("rpsStats"),
		q:         q,
	}

}

// New returns a new cache with no built-in eviction strategy. The cache's name
// is exposed with stats in expvar.
func New(name string) Cache {
	return &cache{
		fetchers:        make(map[reflect.Type]reflect.Value),
		cache:           make(map[interface{}]*CacheEntry),
		cacheSizeExpVar: expvar.NewInt(fmt.Sprintf("cacheSize (%s)", name)),
	}
}

func NewRedamoServer(port int, store store.Store) (*redis.Server, error) {
	redamo := &RedamoHandler{}
	redamo.store = store
	redamo.start = time.Now()
	redamo.tcp = expvar.NewInt("tcp")
	return redis.NewServer(redis.DefaultConfig().Port(port).Handler(redamo))
}

func main() {
	app := cmd.NewAppShell("activity-monitor", "RPC activity monitor")

	app.Action = func(c cmd.Config, stats statsd.Statter, auditlogger *blog.AuditLogger) {
		go cmd.DebugServer(c.ActivityMonitor.DebugAddr)

		amqpConf := c.ActivityMonitor.AMQP
		server, err := rpc.NewMonitorServer(amqpConf, 0, stats)
		cmd.FailOnError(err, "Could not connect to AMQP")

		ae := analysisengine.NewLoggingAnalysisEngine()

		messages := expvar.NewInt("messages")
		server.HandleDeliveries(rpc.DeliveryHandler(func(d amqp.Delivery) {
			messages.Add(1)
			ae.ProcessMessage(d)
		}))

		go cmd.ProfileCmd("AM", stats)

		err = server.Start(amqpConf)
		cmd.FailOnError(err, "Unable to run Activity Monitor")
	}

	app.Run()
}

// NewStats create a new empty stats object
func NewStats() Stats {
	return Stats{
		Status:          "OK",
		EventsReceived:  expvar.NewInt("events_received"),
		EventsSent:      expvar.NewInt("events_sent"),
		EventsIngested:  expvar.NewInt("events_ingested"),
		EventsError:     expvar.NewInt("events_error"),
		EventsDiscarded: expvar.NewInt("events_discarded"),
		QueueSize:       expvar.NewInt("queue_size"),
		QueueMaxSize:    expvar.NewInt("queue_max_size"),
		Clients:         expvar.NewInt("clients"),
		Connections:     expvar.NewInt("connections"),
	}
}

func NewSubMetrics(gw *Gateway) *subMetrics {
	this := &subMetrics{
		gw:          gw,
		ConsumeMap:  make(map[string]metrics.Counter),
		ConsumedMap: make(map[string]metrics.Counter),
		SubQps:      metrics.NewRegisteredMeter("sub.qps", metrics.DefaultRegistry),
		SubTryQps:   metrics.NewRegisteredMeter("sub.try.qps", metrics.DefaultRegistry),
		ClientError: metrics.NewRegisteredMeter(("sub.clienterr"), metrics.DefaultRegistry),
	}

	if Options.DebugHttpAddr != "" {
		this.expConsumeOk = expvar.NewInt("ConsumeOk")
		this.expActiveConns = expvar.NewInt("ConsumeConns")       // TODO
		this.expActiveUpstream = expvar.NewInt("ConsumeUpstream") // TODO
	}

	return this
}

func TestSnapshotExpvars(t *testing.T) {
	test := expvar.NewInt("test")
	test.Add(42)

	vals := map[string]int64{}
	snapshotExpvars(vals)

	assert.Equal(t, vals["test"], int64(42))
}

func init() {
	currentTar = expvar.NewString("CurrentTar")
	tarBytesRead = expvar.NewInt("TarBytesRead")
	tarsFailed = expvar.NewInt("TarsFailed")
	nestedTarsFailed = expvar.NewInt("NestedTarsFailed")
	tarsIndexed = expvar.NewInt("TarsIndexed")
	nestedTarsIndexed = expvar.NewInt("NestedTarsIndexed")
	tarsSkipped = expvar.NewInt("TarsSkipped")
	logsFailed = expvar.NewInt("TracesFailed")
	logsIndexed = expvar.NewInt("TracesIndexed")
}

func TestPublish(t *testing.T) {

	// setup
	port := 2003
	mock := NewMockGraphite(t, port)
	d := 25 * time.Millisecond
	attempts, maxAttempts := 0, 3
	var g *Graphite
	for {
		attempts++
		var err error
		g, err = NewGraphite(fmt.Sprintf("localhost:%d", port), d, d)
		if err == nil || attempts > maxAttempts {
			break
		}
		t.Logf("(%d/%d) %s", attempts, maxAttempts, err)
		time.Sleep(d)
	}
	if g == nil {
		t.Fatalf("Mock Graphite server never came up")
	}

	// register, wait, check
	i := expvar.NewInt("i")
	i.Set(34)
	g.Register("test.foo.i", i)

	time.Sleep(2 * d)
	count := mock.Count()
	if !(0 < count && count <= 2) {
		t.Errorf("expected 0 < publishes <= 2, got %d", count)
	}
	t.Logf("after %s, count=%d", 2*d, count)

	time.Sleep(2 * d)
	count = mock.Count()
	if !(1 < count && count <= 4) {
		t.Errorf("expected 1 < publishes <= 4, got %d", count)
	}
	t.Logf("after second %s, count=%d", 2*d, count)

	// teardown
	ok := make(chan bool)
	go func() {
		g.Shutdown()
		mock.Shutdown()
		ok <- true
	}()
	select {
	case <-ok:
		t.Logf("shutdown OK")
	case <-time.After(d):
		t.Errorf("timeout during shutdown")
	}

}

// Monitoring: export variables via an HTTP handler registered at /debug/vars (http://localhost:8080/debug/vars)
func main() {
	count := expvar.NewInt("count")
	go func() {
		for {
			count.Add(1)
			time.Sleep(time.Second)
		}
	}()
	log.Fatal(http.ListenAndServe("localhost:8080", nil))
}

func main() {
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)

	go collector_updateInfo(ch_update_info_in)
	go collector_stat(ch_ws_counter)
	go queue_reader(ch_update_info_in)

	http.HandleFunc("/ccus", wsHandler)

	g_dt_count_updates = expvar.NewInt("dt_count_updates")
	g_count_updates = expvar.NewInt("count_updates")
	g_count_ws_sessions = expvar.NewInt("count_ws_sessions")
	g_count_request = expvar.NewInt("count_request")

	log.Printf("Listen and serve: %s", WS_LISTEN_ADDR)
	if err := http.ListenAndServe(WS_LISTEN_ADDR, nil); err != nil {
		log.Fatal("ERR! listen and serve:", err)
	}
}

func init() {
	flag.DurationVar(&outageThreshold, "outage_threshold", 5*time.Minute, "Trigger an outage when the duration between two pings from a router is longer than this threshold.")
	flag.StringVar(&outputFile, "output_file", "/tmp/bismark-availability.json", "Write avilability to this file in JSON format")
	flag.StringVar(&outputLevelDb, "output_leveldb", "/tmp/bismark-availability-leveldb", "Write avilability to this leveldb")
	flag.StringVar(&cacheDirectory, "cache_dir", "/tmp/bismark-availability-intervals", "Cache avilability intervals in this directory")
	flag.BoolVar(&excludeGatech, "exclude_gatech", false, "Whether to exclude probes from GT addresses.")
	var dateString string
	flag.StringVar(&dateString, "min_date", "2012-04-13", "Calculate intervals starting at this date")
	flag.Parse()

	dateParsed, err := time.Parse("2006-01-02", dateString)
	if err != nil {
		panic(fmt.Errorf("Invalid date %s: %s", dateString, err))
	}
	minDate = dateParsed

	rowsProcessed = expvar.NewInt("RowsProcessed")
	intervalsCreated = expvar.NewInt("IntervalsCreated")
}

func NewWebInterface(addr string, cm *CallManager, rt RoutingTable) *WebInterface {
	wi := new(WebInterface)
	wi.addr = addr
	wi.cm = cm
	wi.rt = rt
	wi.sm = http.NewServeMux()
	wi.reqcounter = expvar.NewInt("")

	wi.sm.Handle("/", http.HandlerFunc(wi.getDummy()))

	return wi
}