Merge pull request #22 from ispiroglu/tiny-refactors

Tiny refactors

.dockerignore

@@ -0,0 +1,32 @@
+# Include any files or directories that you don't want to be copied to your
+# container here (e.g., local build artifacts, temporary files, etc.).
+#
+# For more help, visit the .dockerignore file reference guide at
+# https://docs.docker.com/engine/reference/builder/#dockerignore-file
+
+**/.DS_Store
+**/.classpath
+**/.dockerignore
+**/.env
+**/.git
+**/.gitignore
+**/.project
+**/.settings
+**/.toolstarget
+**/.vs
+**/.vscode
+**/*.*proj.user
+**/*.dbmdl
+**/*.jfm
+**/bin
+**/charts
+**/docker-compose*
+**/compose*
+**/Dockerfile*
+**/node_modules
+**/npm-debug.log
+**/obj
+**/secrets.dev.yaml
+**/values.dev.yaml
+LICENSE
+README.md

Dockerfile

@@ -1,9 +1,41 @@
-FROM golang:alpine3.18 as build
-WORKDIR /go/src/app
-COPY . .
-RUN mkdir /go/bin/app
-RUN go build -o /go/bin/app/ -v ./cmd/mercurius
-
-FROM alpine:3.18
-COPY --from=build /go/bin/app /app
-ENTRYPOINT ["/app/mercurius"]
+ARG GO_VERSION=1.21
+FROM golang:${GO_VERSION} AS build
+WORKDIR /src
+
+RUN --mount=type=cache,target=/go/pkg/mod/ \
+    --mount=type=bind,source=go.sum,target=go.sum \
+    --mount=type=bind,source=go.mod,target=go.mod \
+    go mod download -x
+
+RUN --mount=type=cache,target=/go/pkg/mod/ \
+    --mount=type=bind,target=. \
+    CGO_ENABLED=0 go build -o /bin/server ./cmd/mercurius/
+
+FROM alpine:latest AS final
+
+RUN --mount=type=cache,target=/var/cache/apk \
+    apk --update add \
+        ca-certificates \
+        tzdata \
+        && \
+        update-ca-certificates
+
+ARG UID=10001
+RUN adduser \
+    --disabled-password \
+    --gecos "" \
+    --home "/nonexistent" \
+    --shell "/sbin/nologin" \
+    --no-create-home \
+    --uid "${UID}" \
+    appuser
+USER appuser
+
+COPY --from=build /bin/server /bin/
+
+EXPOSE 9000
+EXPOSE 6060
+EXPOSE 8081
+EXPOSE 8080
+
+ENTRYPOINT [ "/bin/server" ]

cmd/mercurius-client/const.go

@@ -0,0 +1,15 @@
+package client_example
+
+import "time"
+
+const PublisherCount = 1
+const PublishCount = 100
+
+var TotalPublishCount = uint64(PublisherCount * PublishCount)
+
+const SubscriberCount = 1
+
+var TotalReceiveCount = TotalPublishCount * SubscriberCount
+var StartTime time.Time
+
+const StreamPerSubscriber int = 5

cmd/mercurius-client/publisher-client/mercurius-client.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/google/uuid"
 
+	client_example "github.com/ispiroglu/mercurius/cmd/mercurius-client"
 	logger2 "github.com/ispiroglu/mercurius/internal/logger"
 	"github.com/ispiroglu/mercurius/pkg/client"
 	"go.uber.org/zap"
@@ -21,44 +22,53 @@ const CLIENT_NAME = "Sample Client"
 
 var logger = logger2.NewLogger()
 var messageCount = atomic.Uint64{}
-var N = 100 * 100
+
 var start time.Time
 
 func main() {
-	id, _ := uuid.NewUUID()
-	c, err := client.NewClient(id, ADDR)
-	if err != nil {
-		logger.Error("Err", zap.Error(err))
-	}
 
 	logger.Info("Published Event")
 	var z time.Duration
+	_ = z
 	wg := sync.WaitGroup{}
-	wg.Add(N)
-	uintN := uint64(N)
-	for i := 0; i < N; i++ {
-		go func(w *sync.WaitGroup) {
-			for j := 0; j < 1; j++ {
+	wg.Add(client_example.PublisherCount)
+	client_example.StartTime = time.Now()
+	fmt.Println(client_example.StartTime)
+
+	signal := make(chan struct{})
+	for i := 0; i < client_example.PublisherCount; i++ {
+		id, _ := uuid.NewUUID()
+		c, err := client.NewClient(id, ADDR)
+		if err != nil {
+			logger.Error("Err", zap.Error(err))
+		}
+
+		go func(w *sync.WaitGroup, ch chan struct{}, cl *client.Client) {
+			for j := 0; j < client_example.PublishCount; j++ {
+				<-ch
 				x := messageCount.Add(1)
-				if err := c.Publish(TopicName, []byte(strconv.FormatUint(x, 10)), context.Background()); err != nil {
-					logger.Error("Err", zap.Error(err))
-				}
 				if x == 1 {
 					start = time.Now()
 				}
-				fmt.Println(x)
-				if x == uintN {
+
+				if err := cl.Publish(TopicName, []byte(strconv.FormatUint(x, 10)), context.Background()); err != nil {
+					logger.Error("Err", zap.Error(err))
+					panic(err)
+				}
+
+				fmt.Println("event sent: ", x)
+				if x == client_example.TotalPublishCount {
 					z = time.Since(start)
 				}
-				fmt.Println(strconv.FormatUint(x, 10))
-				//time.Sleep(time.Millisecond)
 			}
 			w.Done()
-		}(&wg)
+		}(&wg, signal, c)
 	}
 
+	time.Sleep(1 * time.Second)
+	close(signal)
 	wg.Wait()
-	fmt.Println("Execution time: ", z)
+	// fmt.Println("Execution time: ", z)
 
 }
 

cmd/mercurius-client/subscriber-client/mercurius-client-sub.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/google/uuid"
 
+	client_example "github.com/ispiroglu/mercurius/cmd/mercurius-client"
 	k "github.com/ispiroglu/mercurius/internal/logger"
 	"github.com/ispiroglu/mercurius/pkg/client"
 	"github.com/ispiroglu/mercurius/proto"
@@ -16,9 +17,6 @@ import (
 
 const ADDR = "0.0.0.0:9000"
 const TopicName = "one-to-one"
-const CLIENT_NAME = "Sample Client"
-const subCount = 100
-const N = 100 * 100 * subCount
 
 var messageCount = atomic.Uint64{}
 var start = time.Time{}
@@ -28,7 +26,7 @@ var ch = make(chan struct{})
 
 func main() {
 
-	for i := 0; i < subCount; i++ {
+	for i := 0; i < client_example.SubscriberCount; i++ {
 		go func() {
 			id, _ := uuid.NewUUID()
 			c, err := client.NewClient(id, ADDR)
@@ -50,11 +48,7 @@ func handler(e *proto.Event) error {
 	if x == 1 {
 		start = time.Now()
 	}
-	// fmt.Println(string(e.Body))
-	if x == N {
-		z := time.Since(start)
-		fmt.Println("Execution time: ", z)
-		ch <- struct{}{}
-	}
+	fmt.Println("event received: ", string(e.Body))
+
 	return nil
 }

cmd/mercurius/mercurius.go

@@ -1,14 +1,17 @@
 package main
 
 import (
+	"encoding/json"
+	"fmt"
 	"net"
-
 	"net/http"
+	_ "net/http/pprof"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"github.com/ispiroglu/mercurius/internal/logger"
 	sv "github.com/ispiroglu/mercurius/internal/server"
 	"github.com/ispiroglu/mercurius/proto"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 )
@@ -19,6 +22,9 @@ const TCP = "tcp"
 var log = logger.NewLogger()
 
 func main() {
+	go func() {
+		fmt.Println(http.ListenAndServe("localhost:6060", nil))
+	}()
 	list, err := net.Listen(TCP, ADDR)
 	if err != nil {
 		log.Fatal("Cannot listen", zap.String("TCP", TCP), zap.String("ADDR", ADDR), zap.Error(err))
@@ -34,6 +40,16 @@ func main() {
 		http.Handle("/metrics", promhttp.Handler())
 		_ = http.ListenAndServe(":8081", nil)
 	}()
+
+	go func() {
+		http.HandleFunc("/topics", func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			_ = json.NewEncoder(w).Encode(server.GetTopics())
+		})
+		_ = http.ListenAndServe(":8080", nil)
+	}()
+
 	if err := grpcServer.Serve(list); err != nil {
 		log.Fatal("Failed to serve", zap.Error(err))
 	}

compose.yaml

@@ -0,0 +1,19 @@
+services:
+  server:
+    build:
+      context: .
+      target: final
+    ports:
+      - 9000:9000
+      - 6060:6060
+      - 8081:8081
+      - 8080:8080
+  prometheus:
+    image: prom/prometheus
+    ports:
+      - 9090:9090
+    command:
+      - --web.enable-remote-write-receiver
+      - --enable-feature=exemplar-storage
+    volumes:
+      - ./prometheus/prometheus.yml:/prometheus/prometheus.yml

go.mod

@@ -11,13 +11,17 @@ require (
 	google.golang.org/protobuf v1.30.0
 )
 
+require github.com/alitto/pond v1.8.3 // indirect
+
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/panjf2000/ants v1.3.0
+	github.com/panjf2000/ants/v2 v2.9.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect

go.sum

@@ -1,3 +1,5 @@
+github.com/alitto/pond v1.8.3 h1:ydIqygCLVPqIX/USe5EaV/aSRXTRXDEI9JwuDdu+/xs=
+github.com/alitto/pond v1.8.3/go.mod h1:CmvIIGd5jKLasGI3D87qDkQxjzChdKMmnXMg3fG6M6Q=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
@@ -21,6 +23,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/panjf2000/ants v1.3.0 h1:8pQ+8leaLc9lys2viEEr8md0U4RN6uOSUCE9bOYjQ9M=
+github.com/panjf2000/ants v1.3.0/go.mod h1:AaACblRPzq35m1g3enqYcxspbbiOJJYaxU2wMpm1cXY=
+github.com/panjf2000/ants/v2 v2.9.0 h1:SztCLkVxBRigbg+vt0S5QvF5vxAbxbKt09/YfAJ0tEo=
+github.com/panjf2000/ants/v2 v2.9.0/go.mod h1:7ZxyxsqE4vvW0M7LSD8aI3cKwgFhBHbxnlN8mDqHa1I=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -35,7 +41,12 @@ github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPH
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -49,6 +60,7 @@ go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
 golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
@@ -64,5 +76,6 @@ google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cn
 google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

go.work

@@ -0,0 +1,6 @@
+go 1.21.5
+
+use (
+	.
+	./test/kafka-cmp
+)