<commit_before>// +build !fasthttp

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/funkygao/gafka/cmd/kateway/meta"
	"github.com/funkygao/gafka/cmd/kateway/meta/zkmeta"
	"github.com/funkygao/gafka/cmd/kateway/store"
	"github.com/funkygao/gafka/cmd/kateway/store/kafka"
	"github.com/funkygao/gafka/ctx"
	"github.com/funkygao/gafka/mpool"
	log "github.com/funkygao/log4go"
	"github.com/gorilla/mux"
	"github.com/julienschmidt/httprouter"
)

type neverEnding byte

var gw *Gateway

func init() {
	log.AddFilter("stdout", log.ERROR, log.NewConsoleLogWriter())
}

func (b neverEnding) Read(p []byte) (n int, err error) {
	if len(p) < 16 {
		for i := range p {
			p[i] = byte(b)
		}
	} else {
		b.Read(p[:len(p)/2])
		copy(p[len(p)/2:], p)
	}
	return len(p), nil
}

func BenchmarkNeverending(b *testing.B) {
	buf := make([]byte, 4096)
	A := neverEnding('A')
	for i := 0; i < b.N; i++ {
		A.Read(buf)
	}
}

func newGatewayForTest(b *testing.B, store string) *Gateway {
	options.zone = "local"
	options.pubHttpAddr = ":9191"
	options.subHttpAddr = ":9192"
	options.store = store
	options.debug = false
	options.disableMetrics = true
	options.enableBreaker = false

	ctx.LoadConfig("/etc/kateway.cf")

	gw := NewGateway("1", time.Hour)
	if err := gw.Start(); err != nil {
		b.Fatal(err)
	}

	return gw
}

func runBenchmarkPub(b *testing.B, store string, msgSize int64) {
	if gw == nil {
		gw = newGatewayForTest(b, store)
	}

	b.ReportAllocs()

	httpReqRaw := strings.TrimSpace(fmt.Sprintf(`
POST /topics/foobar/v1 HTTP/1.1
Host: localhost:9191
User-Agent: Go-http-client/1.1
Content-Length: %d
Content-Type: application/x-www-form-urlencoded
Appid: myappid
Pubkey: mypubkey
Accept-Encoding: gzip`, msgSize)) + "\r\n\r\n"
	b.SetBytes(msgSize + int64(len(httpReqRaw)))

	req, err := http.ReadRequest(bufio.NewReader(strings.NewReader(httpReqRaw)))
	if err != nil {
		b.Fatal(err)
	}

	rw := httptest.NewRecorder()
	lr := io.LimitReader(neverEnding('a'), msgSize)
	body := ioutil.NopCloser(lr)

	param := httprouter.Params{
		httprouter.Param{Key: "topic", Value: "foobar"},
		httprouter.Param{Key: "ver", Value: "v1"},
	}

	for i := 0; i < b.N; i++ {
		rw.Body.Reset()
		lr.(*io.LimitedReader).N = msgSize
		req.Body = body
		gw.pubHandler(rw, req, param)
	}
}

func BenchmarkDirectKafkaProduce1K(b *testing.B) {
	b.Skip("skipped")

	msgSize := 1 << 10
	b.ReportAllocs()
	b.SetBytes(int64(msgSize))

	ctx.LoadConfig("/etc/kateway.cf")
	cf := zkmeta.DefaultConfig("local")
	cf.Refresh = time.Hour
	meta.Default = zkmeta.New(cf)
	meta.Default.Start()
	var wg sync.WaitGroup
	store.DefaultPubStore = kafka.NewPubStore(&wg, false)
	store.DefaultPubStore.Start()

	data := []byte(strings.Repeat("X", msgSize))
	for i := 0; i < b.N; i++ {
		store.DefaultPubStore.SyncPub("me", "foobar", "", data)
	}
}

func BenchmarkKatewayPubKafka(b *testing.B) {
	b.Skip("skipped")

	runBenchmarkPub(b, "kafka", 1<<10)
}

func BenchmarkKatewayPubDumb1K(b *testing.B) {
	runBenchmarkPub(b, "dumb", 1<<10)
}

func BenchmarkGorillaMux(b *testing.B) {
	b.Skip("skip for now")

	router := mux.NewRouter()
	handler := func(w http.ResponseWriter, r *http.Request) {}
	router.HandleFunc("/topics/{topic}/{ver}", handler)
	router.HandleFunc("/ws/topics/{topic}/{ver}", handler)
	router.HandleFunc("/ver", handler)
	router.HandleFunc("/help", handler)
	router.HandleFunc("/stat", handler)
	router.HandleFunc("/ping", handler)
	router.HandleFunc("/clusters", handler)
<commit_msg>fix a unit test compile error<commit_after>// +build !fasthttp

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/funkygao/gafka/cmd/kateway/meta"
	"github.com/funkygao/gafka/cmd/kateway/meta/zkmeta"
	"github.com/funkygao/gafka/cmd/kateway/store"
	"github.com/funkygao/gafka/cmd/kateway/store/kafka"
	"github.com/funkygao/gafka/ctx"
	"github.com/funkygao/gafka/mpool"
	log "github.com/funkygao/log4go"
	"github.com/gorilla/mux"
	"github.com/julienschmidt/httprouter"
)

type neverEnding byte

var gw *Gateway

func init() {
	log.AddFilter("stdout", log.ERROR, log.NewConsoleLogWriter())
}

func (b neverEnding) Read(p []byte) (n int, err error) {
	if len(p) < 16 {
		for i := range p {
			p[i] = byte(b)
		}
	} else {
		b.Read(p[:len(p)/2])
		copy(p[len(p)/2:], p)
	}
	return len(p), nil
}

func BenchmarkNeverending(b *testing.B) {
	buf := make([]byte, 4096)
	A := neverEnding('A')
	for i := 0; i < b.N; i++ {
		A.Read(buf)
	}
}

func newGatewayForTest(b *testing.B, store string) *Gateway {
	options.zone = "local"
	options.pubHttpAddr = ":9191"
	options.subHttpAddr = ":9192"
	options.store = store
	options.debug = false
	options.disableMetrics = true
	options.enableBreaker = false

	ctx.LoadConfig("/etc/kateway.cf")

	gw := NewGateway("1", time.Hour)
	if err := gw.Start(); err != nil {
		b.Fatal(err)
	}

	return gw
}

func runBenchmarkPub(b *testing.B, store string, msgSize int64) {
	if gw == nil {
		gw = newGatewayForTest(b, store)
	}

	b.ReportAllocs()

	httpReqRaw := strings.TrimSpace(fmt.Sprintf(`
POST /topics/foobar/v1 HTTP/1.1
Host: localhost:9191
User-Agent: Go-http-client/1.1
Content-Length: %d
Content-Type: application/x-www-form-urlencoded
Appid: myappid
Pubkey: mypubkey
Accept-Encoding: gzip`, msgSize)) + "\r\n\r\n"
	b.SetBytes(msgSize + int64(len(httpReqRaw)))

	req, err := http.ReadRequest(bufio.NewReader(strings.NewReader(httpReqRaw)))
	if err != nil {
		b.Fatal(err)
	}

	rw := httptest.NewRecorder()
	lr := io.LimitReader(neverEnding('a'), msgSize)
	body := ioutil.NopCloser(lr)

	param := httprouter.Params{
		httprouter.Param{Key: "topic", Value: "foobar"},
		httprouter.Param{Key: "ver", Value: "v1"},
	}

	for i := 0; i < b.N; i++ {
		rw.Body.Reset()
		lr.(*io.LimitedReader).N = msgSize
		req.Body = body
		gw.pubHandler(rw, req, param)
	}
}

func BenchmarkDirectKafkaProduce1K(b *testing.B) {
	b.Skip("skipped")

	msgSize := 1 << 10
	b.ReportAllocs()
	b.SetBytes(int64(msgSize))

	ctx.LoadConfig("/etc/kateway.cf")
	cf := zkmeta.DefaultConfig("local")
	cf.Refresh = time.Hour
	meta.Default = zkmeta.New(cf)
	meta.Default.Start()
	var wg sync.WaitGroup
	store.DefaultPubStore = kafka.NewPubStore(&wg, false)
	store.DefaultPubStore.Start()

	data := []byte(strings.Repeat("X", msgSize))
	for i := 0; i < b.N; i++ {
		store.DefaultPubStore.SyncPub("me", "foobar", nil, data)
	}
}

func BenchmarkKatewayPubKafka(b *testing.B) {
	b.Skip("skipped")

	runBenchmarkPub(b, "kafka", 1<<10)
}

func BenchmarkKatewayPubDumb1K(b *testing.B) {
	runBenchmarkPub(b, "dumb", 1<<10)
}

func BenchmarkGorillaMux(b *testing.B) {
	b.Skip("skip for now")

	router := mux.NewRouter()
	handler := func(w http.ResponseWriter, r *http.Request) {}
	router.HandleFunc("/topics/{topic}/{ver}", handler)
	router.HandleFunc("/ws/topics/{topic}/{ver}", handler)
	router.HandleFunc("/ver", handler)
	router.HandleFunc("/help", handler)
	router.HandleFunc("/stat", handler)
	router.HandleFunc("/ping", handler)
	router.HandleFunc("/clusters", handler)

{'fraction_non_alphanumeric': 0.17779644407111858, 'fraction_numerical': 0.013439731205375893, 'mean_word_length': 4.38763197586727, 'pattern_counts': {'":': 4, '<': 23, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5184265', 'n_tokens_mistral': 3093, 'n_tokens_neox': 2972, 'n_words': 530}