Creando web server para procesar jobs

Me costo un poco entenderlo, pero ya le agarre la onda!

Unos puntos importantes.

• El Dispacher obtendra una cola, que contedra una cola de cada uno de los workers.
• Una vez que el dispacher reciba un trabajo, esta agarrara de su cola a un worker, y mandara la tarea por ese medio
• El worker, procesara el trabajao y volvera a agregar su cola de trabajo a la cola del dispacher

Tambien cree un repositorio en GitHub, con comentarios y con un CodeTour viendo que hace el codigo paso a paso. Pueden clonar mi repo y visualizarlo instalando la extension: https://github.com/CarlosTrejo2308/platzigoserver

Espero que les sirva para entenderlo mejor!

Una manera mas simple de cumplir la interface de http es retornar una handlerFunc en nuestra funcion, con esto podemos recibir multiples parametros y quedara una sintaxis mas limpia

func RequestHandler(jobQueue chan Job) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request)

y al momento de llamarla en main quedaria de esta manera:

http.HandleFunc("/fib", RequestHandler(jobQueue))

Tambien les compartos unos apuntes que tome, espero ayude:

Este es el código que hay que cambiar para mandar multiples request:

func (d *Dispatcher) Dispatch() {
	for {
		select {
		case job := <-d.JobQueue:
			go func() {
				workerJobQueue := <-d.WorkerPool
				workerJobQueue <- job
			}()
		}
	}
}

No necesitas Postman. Thunder Client es una extensión de VSCode que te permite hacer los request directamente desde el editor.

Review: https://www.youtube.com/watch?v=HZx5X3s_Jl4

Pero vamos a ver, ¿dónde antes se ha explicado REST? ¿Por qué da por sentado que ya sabemos lo que es un servidor o lo que es un POST? Por este tipo de cosas la gente suele ver a la programación como algo súper complicado, porque la enseñan mal.

Nota sobre como funciona el canal de canales chan chan Jobs. Cada worker tiene un loop infinito, al comienzo del loop agrega su canal de trabajos al canal de canales y, espera a que haya algun trabajo nuevo en su cola de trabajos.

Cuando el Dispatcher recibe un trabajo, este toma uno de los canales de trabajo de los canales de canales y le agrega un trabajo.

En resumen, cada worker tiene su propia cola y va tomando trabajos de ahi. Tambien hay una cola principal, cuando llega un trabajo nuevo, simplemente se reenvia a la cola del siguiente worker disponible.

En otro aporte deje un refactor para usar una sola cola de trabajos, obteniendo el mismo resultado.

Hice un refactor para simplificar. Elimino el uso de chan chan Job, haciendo que todos los workers lean de la misma cola de trabajo (la principal). El comportamiento es el mismo.

Les comparto un ejemplo de la salida del programa, dejando varios jobs para calcular Fibonacci de 50.

Worker with id 1 Started
Worker with id 2 Started
Worker with id 3 Started
Worker with id 0 Started
Worker with id 2 Finished with result 12586269025
Worker with id 2 Started
Worker with id 1 Finished with result 12586269025
Worker with id 1 Started
Worker with id 3 Finished with result 12586269025
Worker with id 0 Finished with result 12586269025

package main

import (
	"fmt"
	"log"
	"net/http"
	"strconv"
	"time"
)

type Job struct {
	Name   string
	Delay  time.Duration
	Number int
}

type Worker struct {
	Id       int
	JobQueue chan Job
	QuitChan chan bool
}

func NewWorker(id int, jobQueue chan Job) *Worker {
	return &Worker{
		Id:       id,
		JobQueue: jobQueue,
		QuitChan: make(chan bool),
	}
}

func (w *Worker) Start() {
	go func() {
		for {
			select {
			case job := <-w.JobQueue:
				fmt.Println("Worker with id", w.Id, "Started")
				result := Fibonacci(job.Number)
				time.Sleep(job.Delay)
				fmt.Println("Worker with id", w.Id, "Finished with result", result)
			case <-w.QuitChan:
				fmt.Println("Worker with id", w.Id, "Terminated")
				break
			}
		}
	}()
}

func (w *Worker) Stop() {
	go func() {
		w.QuitChan <- true
	}()
}

type Dispatcher struct {
	MaxWorkers int
	JobQueue   chan Job
}

func NewDispatcher(jobQueue chan Job, maxWorkers int) *Dispatcher {
	return &Dispatcher{
		JobQueue:   jobQueue,
		MaxWorkers: maxWorkers,
	}
}

func (d *Dispatcher) Run() {
	for i := 0; i < d.MaxWorkers; i++ {
		worker := NewWorker(i, d.JobQueue)
		worker.Start()
	}
}

func Fibonacci(n int) int {
	if n < 2 {
		return n
	}

	return Fibonacci(n-1) + Fibonacci(n-2)
}

func RequestHandler(w http.ResponseWriter, r *http.Request, jobQueue chan Job) {
	if r.Method != "POST" {
		w.Header().Set("Allow", "POST")
		w.WriteHeader(http.StatusMethodNotAllowed)
	}

	delay, err := time.ParseDuration(r.FormValue("delay"))
	if err != nil {
		http.Error(w, "Invalid delay", http.StatusBadRequest)
		return
	}

	value, err := strconv.Atoi(r.FormValue("value"))
	if err != nil {
		http.Error(w, "Invalid value", http.StatusBadRequest)
		return
	}

	name := r.FormValue("name")
	if name == "" {
		http.Error(w, fmt.Sprintf("Invalid name %s", name), http.StatusBadRequest)
		return
	}

	job := Job{name, delay, value}
	jobQueue <- job
	w.WriteHeader(http.StatusCreated)
}

func main() {
	const (
		maxWorkers   = 4
		maxQueueSize = 20
		port         = ":8081"
	)

	jobQueue := make(chan Job, maxQueueSize)
	dispatcher := NewDispatcher(jobQueue, maxWorkers)
	dispatcher.Run()

	http.HandleFunc("/fibonacci", func(w http.ResponseWriter, r *http.Request) {
		RequestHandler(w, r, jobQueue)
	})

	err := http.ListenAndServe(port, nil)
	if err != nil {
		log.Fatal("Cannot run server")
	}
}