init

callback.go

+package main
+
+import (
+	"sync"
+
+	pbUpstream "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
+)
+
+type apiQueryTxsForAddr struct {
+	uuid string
+	res  chan pbUpstream.TaskResponse
+}
+
+var ApiQueryTxsByAddrForQueue = make(chan apiQueryTxsForAddr, 1000)
+
+func req(uuid string) chan pbUpstream.TaskResponse {
+
+	//ApiQueryTxsByAddrForQueue
+
+	res := make(chan pbUpstream.TaskResponse)
+
+	ApiQueryTxsByAddrForQueue <- apiQueryTxsForAddr{
+		uuid: uuid,
+		res:  res,
+	}
+
+	return res
+
+}
+
+func res(res pbUpstream.TaskResponse) {
+
+	if v, ok := recordmap.LoadAndDelete(res.TaskUuid); ok {
+		resAsV, ok := v.(chan pbUpstream.TaskResponse)
+		if ok {
+			resAsV <- res
+		}
+	}
+}
+
+var recordmap sync.Map
+
+func recordUUID() {
+	for message := range ApiQueryTxsByAddrForQueue {
+		recordmap.Store(message.uuid, message.res)
+	}
+}
+
+// res := make(chan types.Transactions)
+
+// t.ApiQueryTxsByAddrForQueue <- apiQueryTxsForAddr{
+// 	addr: (metaTypes.Address)(addr),
+// 	res:  res,
+// }
+
+// queueTxs := <-res

curl

+curl https://api.openai.com/v1/images/generations \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $OPENAI_API_KEY" \
+  -d '{
+    "model": "dall-e-3",
+    "prompt": "A cute baby sea otter",
+    "n": 1,
+    "size": "1024x1024"
+  }'
+
+
+
+
+{
+  "created": 1589478378,
+  "data": [
+    {
+      "url": "https://..."
+    },
+    {
+      "url": "https://..."
+    }
+  ]
+}
+
+
+
+
+
+
+curl https://api.openai.com/v1/images/edits \
+  -H "Authorization: Bearer $OPENAI_API_KEY" \
+  -F image="@otter.png" \
+  -F mask="@mask.png" \
+  -F prompt="A cute baby sea otter wearing a beret" \
+  -F n=2 \
+  -F size="1024x1024"
+
+
+
+
+{
+  "created": 1589478378,
+  "data": [
+    {
+      "url": "https://..."
+    },
+    {
+      "url": "https://..."
+    }
+  ]
+}
+
+
+
+
+
+curl https://api.openai.com/v1/images/variations \
+  -H "Authorization: Bearer $OPENAI_API_KEY" \
+  -F image="@otter.png" \
+  -F n=2 \
+  -F size="1024x1024"
+
+
+
+
+{
+  "created": 1589478378,
+  "data": [
+    {
+      "url": "https://..."
+    },
+    {
+      "url": "https://..."
+    }
+  ]
+}
+
+
+
+
+
+
+
+
+
+
+
+
+curl https://api.openai.com/v1/chat/completions \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $OPENAI_API_KEY" \
+  -d '{
+    "model": "gpt-3.5-turbo",
+    "messages": [
+      {
+        "role": "system",
+        "content": "You are a helpful assistant."
+      },
+      {
+        "role": "user",
+        "content": "Hello!"
+      }
+    ]
+  }'
+
+
+
+{
+  "id": "chatcmpl-123",
+  "object": "chat.completion",
+  "created": 1677652288,
+  "model": "gpt-3.5-turbo-0613",
+  "system_fingerprint": "fp_44709d6fcb",
+  "choices": [{
+    "index": 0,
+    "message": {
+      "role": "assistant",
+      "content": "\n\nHello there, how may I assist you today?",
+    },
+    "logprobs": null,
+    "finish_reason": "stop"
+  }],
+  "usage": {
+    "prompt_tokens": 9,
+    "completion_tokens": 12,
+    "total_tokens": 21
+  }
+}
+
+
+
+
+
+
+
+
+
+
+
+
+curl http://192.168.1.220:8000/v1/chat/completions \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $OPENAI_API_KEY" \
+  -d '{
+    "model": "gpt-3.5-turbo",
+    "messages": [
+      {
+        "role": "system",
+        "content": "You are a helpful assistant."
+      },
+      {
+        "role": "user",
+        "content": "Hello!"
+      }
+    ]
+  }'
+
+
+
+
+
+curl http://192.168.1.220:8000/v1/images/generations \
+  -H "Content-Type: application/json" \
+  -H "Authorization: Bearer $OPENAI_API_KEY" \
+  -d '{
+    "model": "dall-e-3",
+    "prompt": "A cute baby sea otter",
+    "n": 1,
+    "size": "1024x1024"
+  }'
+
+
+
+
+{
+  "created": 1589478378,
+  "data": [
+    {
+      "url": "https://..."
+    },
+    {
+      "url": "https://..."
+    }
+  ]
+}
+
+
+{
+    "modelType": "picture",
+    "model": "aipaint",
+    "desc": "dogs",
+    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0aW1lU3RhbXAiOjE3MDQwODUzMjA3MjQsInVzZXJSb2xlIjoiY29tbW9uIiwidXNlcklkIjoiNDMifQ.EozgbLG9qIDU6og4t0RWyZ7Sl2m2PSRrqjRd4QVfkBQ"
+}
+
+curl http://192.168.1.220:8000/v1/images/generations \
+  -H "Content-Type: application/json" \
+  -H "Authorization: C1c9oY1c1ejsLWhFqqVo2eMvww6ZfQ4G" \
+  -d '{
+    "modelType": "picture",
+    "model": "aipaint",
+    "desc": "dogs",
+    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0aW1lU3RhbXAiOjE3MDQwODUzMjA3MjQsInVzZXJSb2xlIjoiY29tbW9uIiwidXNlcklkIjoiNDMifQ.EozgbLG9qIDU6og4t0RWyZ7Sl2m2PSRrqjRd4QVfkBQ"
+}'
+
+
+
+
+
+curl http://192.168.1.220:8000/v1/images/generations \
+  -H "Content-Type: application/json" \
+  -H "Authorization: C1c9oY1c1ejsLWhFqqVo2eMvww6ZfQ4G" \
+  -d '{
+    "model": "dall-e-3",
+    "prompt": "A cute baby sea otter",
+    "n": 1,
+    "size": "1024x1024"
+  }'
+
+
+
+curl http://192.168.1.220:8000/v1/chat/completions \
+  -H "Content-Type: application/json" \
+  -H "Authorization: C1c9oY1c1ejsLWhFqqVo2eMvww6ZfQ4G" \
+  -d '{
+    "model": "gpt-3.5-turbo",
+    "messages": [
+      {
+        "role": "system",
+        "content": "You are a helpful assistant."
+      },
+      {
+        "role": "user",
+        "content": "Hello!"
+      }
+    ]
+  }'
+
+
+
+
+
+
+
+curl http://192.168.1.220:8000/v1/chat/completions \
+  -H "Content-Type: application/json" \
+  -H "Authorization: C1c9oY1c1ejsLWhFqqVo2eMvww6ZfQ4G" \
+  -d '{
+    "modelType": "language",
+    "model": "chat",
+    "desc": "hello",
+    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0aW1lU3RhbXAiOjE3MDQwODUzMjA3MjQsInVzZXJSb2xlIjoiY29tbW9uIiwidXNlcklkIjoiNDMifQ.EozgbLG9qIDU6og4t0RWyZ7Sl2m2PSRrqjRd4QVfkBQ"
+}'
+
+
+
+
+
+
+
+
+https://docs.aigic.ai/text-to-image
+
+
+
+{
+    "modelType": "picture",
+    "model": "aipaint",
+    "desc": "dogs",
+    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0aW1lU3RhbXAiOjE3MDQwODUzMjA3MjQsInVzZXJSb2xlIjoiY29tbW9uIiwidXNlcklkIjoiNDMifQ.EozgbLG9qIDU6og4t0RWyZ7Sl2m2PSRrqjRd4QVfkBQ"
+}
+
+
+{
+    "code": 200,
+    "msg": null,
+    "content": "https://paint4art.oss-cn-beijing.aliyuncs.com/aiimages/SN90P6aKwh.jpg"
+}
+
+
+
+
+
+{
+    "modelType": "language",
+    "model": "chat",
+    "desc": "hello",
+    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0aW1lU3RhbXAiOjE3MDQwODUzMjA3MjQsInVzZXJSb2xlIjoiY29tbW9uIiwidXNlcklkIjoiNDMifQ.EozgbLG9qIDU6og4t0RWyZ7Sl2m2PSRrqjRd4QVfkBQ"
+}
+
+{
+    "code": 200,
+    "msg": null,
+    "content": "Hello! How can I assist you today?"
+}
+
+
+
+
+
+
+
+
+
+curl http://192.168.1.220:6000/images/generations \
+  -H "Content-Type: application/json" \
+  -H "Authorization: C1c9oY1c1ejsLWhFqqVo2eMvww6ZfQ4G" \
+  -d '{
+    "modelType": "picture",
+    "model": "aipaint",
+    "desc": "dogs",
+    "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ0aW1lU3RhbXAiOjE3MDQwODUzMjA3MjQsInVzZXJSb2xlIjoiY29tbW9uIiwidXNlcklkIjoiNDMifQ.EozgbLG9qIDU6og4t0RWyZ7Sl2m2PSRrqjRd4QVfkBQ"
+}'
+

docker-compose.yml

+version: "3.9"
+
+services:
+  ai-api-mgr:
+      image: caduceus/ai-api-mgr:v0.0.5
+      container_name: ai-api-mgr
+      entrypoint:  sh -c 'ai-api-mgr -kafkaBroker=192.168.1.220:9092 -questAddr=192.168.1.220:9009   -callbackIpAddr=192.168.1.10:6001 -mysqlAddr=192.168.1.220 -mysqlDbName=ai -mysqlUser=ai -mysqlPassWd=RFnnKHRar5xk7TEF -redisPassWd=iH0g2CqzjI6SfercGwsT -redisAddr=192.168.1.220:6379'
+      ports:
+         - 6001:6001
\ No newline at end of file

dockerfile

+FROM golang:1.21.6-alpine3.18 AS build
+#FROM golang:1.20.6-alpine3.18 AS build
+#FROM golang:1.21.3 AS base 
+#FROM golang:1.20.6-alpine3.18 AS build
+
+WORKDIR /go/src/github.com/odysseus/
+
+ENV https_proxy   'http://192.168.1.180:1080'
+
+#RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories
+
+RUN mkdir -p ./go-kafka/
+COPY ./  ./go-kafka
+COPY ./payment ./payment
+COPY ./odysseus-protocol ./odysseus-protocol
+
+#RUN  pwd && ls  && sleep 10
+# RUN go mod tidy
+#RUN go build  -o /go-kafka
+
+RUN  cd go-kafka && go mod tidy && go build  -o /ai-api-mgr
+
+
+FROM alpine
+WORKDIR /root
+
+COPY  --from=build /ai-api-mgr /usr/bin/ai-api-mgr
+
+#ENTRYPOINT [ "/usr/bin/ai-api-mgr" ]
+
+
+
+
+
+# FROM alpine
+# WORKDIR /root
+
+# COPY  --from=base /ai-api-mgr /usr/bin/
+# RUN ls /usr/bin/ai-api-mgr && sleep 30
+
+# ENTRYPOINT [ "/usr/bin/ai-api-mgr" ]
+
+# RUN ls /ai-api-mgr && sleep 30
+# FROM alpine
+# WORKDIR /root
+# RUN ls /usr/bin/ && sleep 10
+# COPY  --from=base /ai-api-mgr /usr/bin/
+# RUN ls /usr/bin/ai-api-mgr && sleep 10
+
+
+
+#/usr/bin/go-kafka
+# Add entrypoint script
+#COPY ./scripts/entrypoint.sh /usr/local/bin/entrypoint.sh
+#RUN chmod u+x /usr/local/bin/entrypoint.sh

docs/docs.go

+// Package docs Code generated by swaggo/swag. DO NOT EDIT
+package docs
+
+import "github.com/swaggo/swag"
+
+const docTemplate = `{
+    "schemes": {{ marshal .Schemes }},
+    "swagger": "2.0",
+    "info": {
+        "description": "{{escape .Description}}",
+        "title": "{{.Title}}",
+        "contact": {},
+        "version": "{{.Version}}"
+    },
+    "host": "{{.Host}}",
+    "basePath": "{{.BasePath}}",
+    "paths": {}
+}`
+
+// SwaggerInfo holds exported Swagger Info so clients can modify it
+var SwaggerInfo = &swag.Spec{
+	Version:          "",
+	Host:             "",
+	BasePath:         "",
+	Schemes:          []string{},
+	Title:            "",
+	Description:      "",
+	InfoInstanceName: "swagger",
+	SwaggerTemplate:  docTemplate,
+	LeftDelim:        "{{",
+	RightDelim:       "}}",
+}
+
+func init() {
+	swag.Register(SwaggerInfo.InstanceName(), SwaggerInfo)
+}

docs/swagger.json

+{
+    "swagger": "2.0",
+    "info": {
+      "title": "GPU+Model service",
+      "version": "1.0.0"
+    },
+    "paths": {
+      "/{TaskType}": {
+        "post": {
+          "summary": "customer private api",
+          "operationId": "place a task",
+
+          "parameters": [
+            {
+                "name":"TaskType",
+                "in": "path",
+                "required": true,
+                "type": "string",
+                "default": "startdocker"
+
+            },
+            {
+              "name": "apikey",
+              "in": "header",
+              "required": true,
+              "type": "string",
+              "maxLength": 32,
+              "default": "YNExx9qR8aBoZuxfNTBCDkVJ3EiedIma"
+            },
+            {
+                "in": "body",  
+                "schema":{
+                    "$ref": "#/definitions/ReqTask"
+                },
+                "examples":{
+                    "modelType":{
+                        "value":"language"
+                    },
+                    "model":{
+                        "value":"chat"
+                    }
+                }
+            }
+          ],
+          "responses": {
+            "200": {
+              "description": "success",
+              "schema": {
+                "$ref": "#/definitions/User"
+              }
+            }
+          }
+        }
+      },
+      "/callback": {
+        "post": {}
+      }
+    },
+    "definitions": {
+       "ReqTask":{
+        "type": "object",
+        "properties": {
+            "modelType": {
+                "type": "string"
+            },
+            "model":{
+                "type": "string"
+            },
+            "desc":{
+                "type": "string"
+            },
+            "token":{
+                "type": "string"
+            }
+        }
+       },
+
+      "User": {
+        "type": "object",
+        "properties": {
+          "id": {
+            "type": "integer",
+            "format": "int64"
+          },
+          "username": {
+            "type": "string"
+          },
+          "email": {
+            "type": "string"
+          }
+        }
+      },
+      "Error": {
+        "type": "object",
+        "properties": {
+          "code": {
+            "type": "integer",
+            "format": "int32"
+          },
+          "message": {
+            "type": "string"
+          }
+        }
+      }
+    }
+  }
+
+

docs/swagger.yaml

+info:
+  contact: {}
+paths: {}
+swagger: "2.0"

go.mod

+module github.com/odysseus/go-kafka
+
+go 1.21.3
+
+replace github.com/odysseus/odysseus-protocol => ../odysseus-protocol
+
+replace github.com/odysseus/payment => ../payment
+
+require (
+	github.com/IBM/sarama v1.42.1
+	github.com/gofiber/contrib/swagger v1.1.1
+	github.com/gofiber/fiber/v2 v2.52.0
+	github.com/gogo/protobuf v1.3.2
+	github.com/odysseus/odysseus-protocol v0.0.0-00010101000000-000000000000
+	github.com/odysseus/payment v0.0.0-00010101000000-000000000000
+	github.com/questdb/go-questdb-client/v2 v2.0.0
+	github.com/swaggo/swag v1.16.3
+)
+
+require (
+	github.com/KyleBanks/depth v1.2.1 // indirect
+	github.com/andybalholm/brotli v1.0.5 // indirect
+	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
+	github.com/astaxie/beego v1.12.3 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/eapache/go-resiliency v1.4.0 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
+	github.com/eapache/queue v1.1.0 // indirect
+	github.com/go-openapi/analysis v0.21.4 // indirect
+	github.com/go-openapi/errors v0.20.3 // indirect
+	github.com/go-openapi/jsonpointer v0.19.5 // indirect
+	github.com/go-openapi/jsonreference v0.20.0 // indirect
+	github.com/go-openapi/loads v0.21.2 // indirect
+	github.com/go-openapi/runtime v0.26.0 // indirect
+	github.com/go-openapi/spec v0.20.8 // indirect
+	github.com/go-openapi/strfmt v0.21.7 // indirect
+	github.com/go-openapi/swag v0.22.3 // indirect
+	github.com/go-openapi/validate v0.22.1 // indirect
+	github.com/go-sql-driver/mysql v1.7.1 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/gomodule/redigo v2.0.0+incompatible // indirect
+	github.com/google/uuid v1.5.0 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/hashicorp/golang-lru v0.5.4 // indirect
+	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
+	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
+	github.com/jcmturner/gofork v1.7.6 // indirect
+	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
+	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/klauspost/compress v1.17.0 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mattn/go-runewidth v0.0.15 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/oklog/ulid v1.3.1 // indirect
+	github.com/pierrec/lz4/v4 v4.1.18 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/redis/go-redis/v9 v9.4.0 // indirect
+	github.com/rivo/uniseg v0.2.0 // indirect
+	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/valyala/bytebufferpool v1.0.0 // indirect
+	github.com/valyala/fasthttp v1.51.0 // indirect
+	github.com/valyala/tcplisten v1.0.0 // indirect
+	go.mongodb.org/mongo-driver v1.11.3 // indirect
+	golang.org/x/crypto v0.14.0 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sys v0.15.0 // indirect
+	golang.org/x/tools v0.14.0 // indirect
+	google.golang.org/protobuf v1.32.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

makefile

+.PHONY: default metaring all clean fmt docker
+
+VERSION := $(shell echo $(shell git describe --tags))
+
+
+docker:
+	docker build -t caduceus/ai-api-mgr:${VERSION} .
+

readme.md

+# Go Web Server with Kafka Producer and Consumer using Fiber
+
+This is a simple Go web server that implements a Kafka producer and consumer using the Fiber web framework. The server has two endpoints: one for sending data to the Kafka producer and another for receiving data from the Kafka consumer.
+
+## Prerequisites
+
+Before running the application, you need to have the following installed:
+- Golang (https://go.dev/dl/)
+- Kafka (https://kafka.apache.org/downloads)
+- Zookeeper (https://zookeeper.apache.org/releases.html)
+
+## Setup
+
+Clone the repository:
+
+```
+git clone https://github.com/Sahil-4555/go-kafka.git
+```
+
+Install the required dependencies:
+
+```
+go mod tidy
+```
+
+Make sure you have Kafka running on `localhost:9092` or update the `kafkaBrokers` variable in `main.go` with your Kafka broker addresses.
+
+## Start ZooKeeper
+
+Apache Kafka relies on ZooKeeper for maintaining metadata. You need to start ZooKeeper before starting Kafka.
+
+Open a terminal in the folder where Kafka is downloaded, and then run ZooKeeper using the following command:
+
+```
+.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
+```
+
+## Start Kafka Broker
+
+Open another terminal in the folder where kafka is downloaded, and start the kafka server using the following command:
+
+```
+.\bin\windows\kafka-server-start.bat .\config\server.properties
+```
+
+If You want to consume and print messages from the Kafka topic in real-time, starting from the beginning of the topic, while connecting to the Kafka broker running on localhost:9092. run the following command in another terminal:
+
+```
+.\bin\windows\kafka-console-consumer.bat --topic <topic name> --bootstrap-server localhost:9092 --from-beginning
+```
+
+## Running the application
+
+To run the application, execute the following command:
+
+```
+go run main.go
+```
+
+The server will start listening on `http://localhost:3000`.
+
+## Endpoints
+
+To send data to the Kafka producer, make a GET request to the `/producer` endpoint with a message as a parameter.
+
+```
+// GET METHOD
+http://localhost:3000/producer/:message
+```
+
+To receive data from the Kafka consumer, make a GET request to the `/consumer` endpoint. The endpoint will respond with the last received message from the Kafka consumer or a default message if no message is available within 4 seconds.
+
+```
+// GET METHOD
+http://localhost:3000/consumer
+```
+
+## How it Works
+
+The application uses the Fiber web framework to handle HTTP requests. When a client sends a GET request to the `/producer` endpoint with a message, the message is sent to the Kafka producer via the `producerMessages` channel.
+
+The Kafka producer, running in a separate goroutine, reads messages from the `producerMessages` channel and sends them to the Kafka topic. The producer introduces a random delay between 1 to 3 seconds for message push to simulate real-world scenarios.
+
+The Kafka consumer, also running in a separate goroutine, continuously listens for new messages from the Kafka topic. When a new message is received, it is added to the internal `messages` slice using a mutex to ensure concurrent-safe access.
+
+The `/consumer` endpoint, upon receiving a GET request, checks the `consumerMessages` channel for any available message within 4 seconds. If a message is available, it is sent as a response. Otherwise, a default message is sent indicating that no messages are available at the moment.
+
+The global counter `counter` is used to keep track of the messages sent to the Kafka producer.
+
+
+
+
+
+
+