import time
import random

def unreliable_function():
    if random.random() < 0.7:
        raise Exception("an error occurred")
    return "OK"

def retry(func, retries=3, delay=2):
    for attempt in range(retries):
        try:
            result = func()
            return result
        except Exception as e:
            print(f"try again. retry time: %d" % retries)
            time.sleep(delay)
    raise Exception("unexpected error")

# Retry Pattern
result = retry(unreliable_function)
print(result)

func HandleRequest(ctx context.Context, db *sql.DB) {
    query := "SELECT * FROM users"
    rows, err := db.QueryContext(ctx, query)
    if err != nil {
        log.Fatal(err)
    }
    defer rows.Close()

    for rows.Next() {
        // پردازش داده‌ها
    }
}

syntax = "proto3";

package user;

message User {
    string name = 1;
    int32 age = 2;
}

go get google.golang.org/protobuf/cmd/protoc-gen-go

protoc --go_out=. user.proto

package main

import (
    "log"
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "google.golang.org/protobuf/proto"
    "your_project/user"
)

func main() {
    producer, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost:9092"})
    if err != nil {
        log.Fatal(err)
    }
    defer producer.Close()

    // create user
    user := &user.User{
        Name: "Alice",
        Age:  30,
    }

    // serialize data
    data, err := proto.Marshal(user)
    if err != nil {
        log.Fatal("Failed to serialize user:", err)
    }

    // publish a message
    topic := "users"
    err = producer.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
        Value:          data,
    }, nil)

    if err != nil {
        log.Fatal("Failed to send message:", err)
    }

    producer.Flush(15 * 1000)
    log.Println("User sent to Kafka:", user.Name)
}

package main

import (
    "log"
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "google.golang.org/protobuf/proto"
    "your_project/user"
)

func main() {
    consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "group.id":          "user_group",
        "auto.offset.reset": "earliest",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer consumer.Close()

    consumer.SubscribeTopics([]string{"users"}, nil)

    for {
        msg, err := consumer.ReadMessage(-1)
        if err != nil {
            log.Println("Error reading message:", err)
            continue
        }

        // deserialization
        receivedUser := &user.User{}
        if err := proto.Unmarshal(msg.Value, receivedUser); err != nil {
            log.Println("Failed to unmarshal user:", err)
            continue
        }

        // use info
        log.Printf("Received User: Name: %s, Age: %d\n", receivedUser.Name, receivedUser.Age)
    }
}

docker-compose up -d

pip install redis

import json

from redis import StrictRedis


server = StrictRedis(host="localhost", port=6399, password="redis_password", db=0)

# redis ping
print(server.ping())

topic = "example_topic"

data = {
    "name": "alireza",
    "age": 22,
}

server.publish(channel=topic, message=json.dumps(data))

import json

from redis import StrictRedis


client = StrictRedis(host="localhost", port=6399, password="redis_password", db=0)

topic = "example_topic"

pubsub = client.pubsub()

pubsub.subscribe(topic)

print("waiting for message...")

while True:
    for message in pubsub.listen():
        if message["data"] == 1:
            continue
        match message["type"]:
            case topic:
                # TODO - change serialization. json is not good
                data = json.loads(message["data"])
                print("received message", data["name"], data["age"])

{"alg": "HS256"}

{"user_id": 1, "exp": 111222558}

   header.payload
   

   header.payload.signature

package main

import (
"fmt"
"net/http"
"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
  CheckOrigin: func(r *http.Request) bool {
    // checking conditions
    return true
  },
}

func handleConnections(w http.ResponseWriter, r *http.Request) {
  // upgrade http request to websocket
  ws, err := upgrader.Upgrade(w, r, nil)
  if err != nil {
    fmt.Println(err)
    return
  }
  defer ws.Close()

  // messages
  for {
    messageType, msg, err := ws.ReadMessage()
    if err != nil {
      fmt.Println(err)
      break
    }
    fmt.Printf("Received: %s\n", msg)

    err = ws.WriteMessage(messageType, msg)
    if err != nil {
      fmt.Println(err)
      break
    }
  }
}

func main() {
  http.HandleFunc("/", handleConnections)
  fmt.Println("Server started on :8080")
  err := http.ListenAndServe(":8080", nil)
  if err != nil {
    fmt.Println("Error starting server:", err)
  }
}