Как отправить данные при помощи канала в определённую Go рутину?

Question

[Владимир Грабко]

У меня есть куча рутин (колбек http роутера). И мне необходимо с другой рутины отправить данные через канал в определённую. Вот код

package main

import (
	"log"
	"net/http"

	"github.com/gorilla/websocket"
	"github.com/julienschmidt/httprouter"
)

func main() {
	go func() {
		//здесь я получаю данные которые необходимо "проксировать"
	}()
	router := httprouter.New()
	router.GET("/events/ws/", EventsWSHandler)
	router.NotFound = http.FileServer(http.Dir("www/"))
	log.Fatal(http.ListenAndServe(":8080", router))

}

var upgrader = websocket.Upgrader{}

func EventsWSHandler(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {

	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Print("upgrade:", err)
		return
	}
	defer c.Close()
	for {
		mt, message, err := c.ReadMessage()
		if err != nil {
			log.Println("read:", err)
			break
		}
		log.Printf("recv: %s", message)
		//а здесь я их отправляю юзеру
		err = c.WriteMessage(mt, message)
		if err != nil {
			log.Println("write:", err)
			break
		}
	}
}

Если бы у меня был обычный tcp сервер то я бы это сделал при помощи пула клиентов (спасибо вам что научили меня его делать) но это же http. Я понятия не имею как сделать тут пул

Comments

[Александр Павлюк]

Абсолютно также делается пул¸ как в случае с tcp

[Владимир Грабко]

Александр Павлюк: ну... Я там делаю пул при запуске новых рутин, а тут я же абстрагирован от этого

[Александр Павлюк]

Владимир Грабко: у вас функция EventsWSHandler вызывается ведь при подключении клиента, вот и добавляйте клиента в пул при удачном апгрейде сокета

[Владимир Грабко]

Александр Павлюк: я так понял мне просто делать пулл коннектов ws, а потом его дёргать из функции в которой получаю. Верно?

[Владимир Грабко]

Александр Павлюк: напишите ответ. Я отметку вам поставлю.

[Александр Павлюк]

Владимир Грабко: я вам пример кода пишу, скоро запощу )))

[Владимир Грабко]

Александр Павлюк: я уже наговнокодил )) Но вам зарание благодарен

Answer 1

[Александр Павлюк]

Грубо говоря, вот так можно

package main

import (
	"log"
	"net/http"
	"sync/atomic"
	"time"

	"github.com/gorilla/websocket"
	"github.com/julienschmidt/httprouter"
)

type Client struct {
	c *websocket.Conn
}

func (this *Client) sendMessage(msg string) {
	err := this.c.WriteMessage(websocket.TextMessage, []byte(msg))
	if err != nil {
		log.Println("write:", err)
	}
}

type Pool struct {
	lastId     uint64
	addChan    chan (*Client)
	removeChan chan (uint64)
	msgChan    chan (string)
	clients    map[uint64]*Client
}

func (this *Pool) serviceRoutine() {
	for {
		select {
		case newClient := <-this.addChan:
			this.clients[this.lastId] = newClient
		case id := <-this.removeChan:
			delete(this.clients, id)
			log.Printf("Client with id %d disconnected\n", id)
		case msg := <-this.msgChan:
			for _, client := range this.clients {
				client.sendMessage(msg)
			}
		}
	}
}

func (this *Pool) addClient(client *Client) uint64 {
	newId := atomic.AddUint64(&this.lastId, 1)
	this.addChan <- client
	log.Printf("Client with id %d connected\n", newId)
	return newId
}

func NewPool() *Pool {
	pool := &Pool{
		addChan:    make(chan *Client),
		removeChan: make(chan uint64),
		msgChan:    make(chan string, 5),
		clients:    make(map[uint64]*Client),
	}
	go pool.serviceRoutine()
	return pool
}

var pool *Pool = NewPool()

func main() {
	go func() {
		//здесь я получаю данные которые необходимо "проксировать"
		ticker := time.NewTicker(time.Second)
		for range ticker.C {
			// Просто посылаем текущее время раз в секунду всем подключенным клиентам
			pool.msgChan <- time.Now().Format("Mon Jan 2 15:04:05 -0700 MST 2006")
		}
	}()
	router := httprouter.New()
	router.GET("/events/ws/", EventsWSHandler)
	router.NotFound = http.FileServer(http.Dir("www/"))
	log.Fatal(http.ListenAndServe(":3000", router))
}

var upgrader = websocket.Upgrader{}

func EventsWSHandler(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {

	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Print("upgrade:", err)
		return
	}
	defer c.Close()

	id := pool.addClient(&Client{c: c})
	defer func() {
		pool.removeChan <- id
	}()

	for {
		mt, message, err := c.ReadMessage()
		if err != nil {
			log.Println("read:", err)
			break
		}
		log.Printf("recv: %d %s", mt, message)
	}
}

Comments

[Владимир Грабко]

огромное спасибо

Answer 2

[qweqwa]

https://habrahabr.ru/post/278349/

Интересные способы использования Go каналов

Я написал этот пост, чтобы задокументировать доклад про Go каналы Джона Грэм-Камминга на конференции GopherCon 2014. Доклад назывался «Краткое руководство по каналам» и он доступен для просмотра на youtube.com.

На протяжении доклада нам представляют интересные способы использования Go каналов и раскрывают возможности и преимущества конкурентного программирования. Лично мне этот доклад открыл глаза на несколько новых способов структурирования программ и новых техник для синхронизации по нескольким ядрам процессора.

Следующие примеры демонстрируют различные техники, как использовать каналы в Go. Код был специально упрощен для их понимания. Не стоит его использовать для продакшн версий. Например, пропущены все обработки ошибок.