Как отправить данные во все рутины?

Question

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

Спасибо FireGM за то что помог мне исправить сделать отправку сообщений с термнала клиентам. По идее код должен отправить данные всем клиентам. Сейчас подключил два клиента, пишу первое сообщение оно приходит только первому клиенту, пишу второе сообщение оно приходит только второму клиенту, пишу третье смс оно приходит только первому и т.д.

Код сервера

package main

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"strconv"
	"strings"
)

var messages chan string

func main() {
	messages = make(chan string, 10)
	var port int = 3333
	listen, err := net.Listen("tcp4", ":"+strconv.Itoa(port))
	defer listen.Close()
	if err != nil {
		log.Fatalf("Прослушивание порта %d не удалось,\r\n\r\n %s", port, err)
		os.Exit(1)
	}

	go func() {

		log.Printf("Сервер слушает порт: %d", port)
		reader := bufio.NewReader(os.Stdin)
		for {
			fmt.Print("Enter text: ")
			text, _ := reader.ReadString('\n')
			log.Print(text)
			messages <- text
		}

	}()
	for {
		conn, err := listen.Accept()
		if err != nil {
			log.Fatalln(err)
			continue
		}
		go handler(conn)
	}
}
func handler(conn net.Conn) {
	defer conn.Close()
	var (
		buf = make([]byte, 1024)
		r   = bufio.NewReader(conn)
		w   = bufio.NewWriter(conn)
	)
ILOOP:
	for {
		n, err := r.Read(buf)
		data := string(buf[:n])
		switch err {
		case io.EOF:
			break ILOOP
		case nil:
			//log.Println("Получил:", data)
			if isTransportOver(data) {
				break ILOOP
			}
		default:
			log.Fatalf("Ошибка при получении:%s", err)
			return
		}
	}
	//данные из канала
	msg := <-messages
	w.Write([]byte(msg))
	w.Flush()

}
func isTransportOver(data string) (over bool) {
	over = strings.HasSuffix(data, "\r\n\r\n")
	return
}

Почему так происходит? зарание спасибо

Comments

[Илья]

Потому что надо делать пул клиентов и отправлять всем в пуле. Здесь саму архитектуру надо переделать.

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

FireGM: poll клиентов это список их ip:port на которые сервер шлёт данные?

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

Владимир Грабко: пул клиентов это массив или map каналов, по одному каналу на клиента. Сообщение, записанное в канал, может быть прочитано из него только один раз! Поэтому у вас оно доходит только одному клиенту.

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

Александр Павлюк: тоесть мне команду в канал подавать n раз?

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

Александр Павлюк: сколько клиентов столько раз отправить запрос

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

Владимир Грабко: если подадите команду n раз, нет гарантии, что каждый клиент получит ее по разу, порядок чтения из канала не гарантировал. Какой клиент быстрее успеет, тот и папа. Нужно именно n каналов.

Answer 1

[Илья]

https://gist.github.com/drewolson/3950226 вот неплохой пример реализации пула клиентов. Просто вырезать лишнее и норм будет.

переделал тот пример, с удалением умерших коннекторов при отправке. Вырезал немного, работает только на отправку клиенту.

package main

import (
	"bufio"
	"fmt"
	"log"
	"net"
	"os"
)

type Client struct {
	writer *bufio.Writer
}

func (client *Client) Write(data string) error {
	fmt.Println("client write")
	_, err := client.writer.WriteString(data)
	err = client.writer.Flush()
	return err
}

func NewClient(connection net.Conn) *Client {
	writer := bufio.NewWriter(connection)
	client := &Client{
		writer: writer,
	}
	return client
}

type ChatRoom struct {
	clients []*Client
}

func (chatRoom *ChatRoom) Broadcast(data string) {
	deleted := 0
	for i := range chatRoom.clients {
		j := i - deleted
		if err := chatRoom.clients[j].Write(data); err != nil {
			fmt.Println(err, "deleted error")
			chatRoom.clients = chatRoom.clients[:j+copy(chatRoom.clients[j:], chatRoom.clients[j+1:])]
			deleted++
		}
	}
}

func (chatRoom *ChatRoom) Join(connection net.Conn) {
	client := NewClient(connection)
	chatRoom.clients = append(chatRoom.clients, client)
}

func NewChatRoom() *ChatRoom {
	chatRoom := &ChatRoom{
		clients: make([]*Client, 0),
	}

	return chatRoom
}

func main() {
	chatRoom := NewChatRoom()

	listener, _ := net.Listen("tcp4", ":3333")
	go readAndSend(chatRoom)

	for {
		conn, err := listener.Accept()
		if err != nil {
			continue
		}
		chatRoom.Join(conn)
	}
}

func readAndSend(chatRoom *ChatRoom) {

	reader := bufio.NewReader(os.Stdin)
	for {
		fmt.Print("Enter text: ")
		text, _ := reader.ReadString('\n')
		log.Print(text)
		chatRoom.Broadcast(text)
	}

}

Comments

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

Что в этом примере произойдет при отключении клиента?

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

Александр Павлюк: там вообще не чего не происходит. запустил пример сервера (как есть) сворганил простой клиент

package main

import (
	"log"
	"net"
	"strconv"
	"strings"
)

func main() {
	Client("46.63.122.27", 6666)
}

func Client(ip string, port int) {
	addr := strings.Join([]string{ip, strconv.Itoa(port)}, ":")
	conn, err := net.Dial("tcp", addr)
	defer conn.Close()
	if err != nil {
		log.Fatalln(err)
	}
	conn.Write([]byte("test"))
	log.Printf("Отправил: %s", "test")
	buff := make([]byte, 1024)
	n, _ := conn.Read(buff)
	log.Printf("Ответ: %s", string(buff[:n]))
}

в консольке пишет Отправил бла бла.
Но а назад не получаю нечего. Более чем уверен что что то не так с клиентом

[Илья]

Владимир Грабко: там же просто пример...

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

Владимир Грабко: используйте для отладки telnet лучше

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

FireGM: Александр Павлюк: а в telnet всё работает

Answer 2

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

Вот пример с пулом клиентов и сохранением в map.
Клиентам рассылается то, что набрано на сервере, данные, пришедшие от клиентов обрабатываются.

package main

import (
	"bufio"
	"fmt"
	"log"
	"net"
	"os"
	"strconv"
	"strings"
)

var messages chan string

type Client struct {
	Id      uint64
	MsgChan chan (string)
}

type ClientPool struct {
	counter    uint64
	joins      chan (Client)
	lefts      chan (uint64)
	ids        chan (uint64)
	broadcasts chan (string)
	clients    map[uint64]Client
}

func NewClientPool() *ClientPool {
	pool := ClientPool{
		joins:      make(chan (Client)),
		lefts:      make(chan (uint64)),
		ids:        make(chan (uint64)),
		broadcasts: make(chan (string)),
		clients:    make(map[uint64]Client),
	}

	go func() {
		var id uint64
		for {
			id++
			pool.ids <- id
		}
	}()

	go func() {
		for {
			select {
			case newClient := <-pool.joins:
				pool.clients[newClient.Id] = newClient
			case idForDeletion := <-pool.lefts:
				delete(pool.clients, idForDeletion)
			case text := <-pool.broadcasts:
				for _, client := range pool.clients {
					client.MsgChan <- text
				}
			}
		}
	}()

	return &pool
}

func (this *ClientPool) NewClient() Client {
	client := Client{
		Id:      <-this.ids,
		MsgChan: make(chan (string)),
	}
	this.joins <- client
	return client
}

func (this *ClientPool) BroadcastMessage(text string) {
	this.broadcasts <- text
}

func (this *ClientPool) DeleteClient(client Client) {
	this.lefts <- client.Id
}

func main() {
	pool := NewClientPool()
	var port int = 3333
	listen, err := net.Listen("tcp4", ":"+strconv.Itoa(port))
	defer listen.Close()
	if err != nil {
		log.Fatalf("Прослушивание порта %d не удалось,\r\n\r\n %s", port, err)
		os.Exit(1)
	}

	go func() {

		log.Printf("Сервер слушает порт: %d", port)
		reader := bufio.NewReader(os.Stdin)
		for {
			fmt.Print("Enter text: ")
			text, _ := reader.ReadString('\n')
			log.Print(text)
			pool.BroadcastMessage(text)
		}

	}()

	for {
		conn, err := listen.Accept()
		if err != nil {
			log.Fatalln(err)
			continue
		}
		go handler(conn, pool)
	}
}

func handler(conn net.Conn, pool *ClientPool) {
	client := pool.NewClient()
	defer pool.DeleteClient(client)
	defer conn.Close()
	defer log.Printf("Клиент с номером %d вышел\n", client.Id)

	log.Printf("Новый клиент присоединен, ему присвоен номер %d", client.Id)

	incomingClientMsg := make(chan string)

	go func() {
		reader := bufio.NewReader(conn)
		for {
			text, err := reader.ReadString('\n')
			if err != nil {
				close(incomingClientMsg)
				return
			}
			incomingClientMsg <- text
		}
	}()

	for {
		select {
		case text := <-incomingClientMsg:
			text = strings.Trim(text, "\r\n")
			if text == "/quit" {
				return
			}
			log.Printf("Клиент с номером %d написал: %s\n", client.Id, text)
		case text := <-client.MsgChan:
			_, err := conn.Write([]byte(text))
			if err != nil {
				return
			}
		}
	}
}

Comments

[Илья]

О, у вас потокобезопасное удаление. У меня нет, но, правда, все равно не упадёт. Хотя и неприятно.

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

Добрый день я уже второй день не могу понять как так выходит что сообщения отправляются всем участникам. Не могли бы вы на пальцах объяснить почему так происходит? Зарание спасибо.

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

Владимир Грабко: хранится список всех подключенных клиентов clients map[uint64]Client
Если в канал broadcasts пришло сообщение, то пробегаемся циклом по этому списку и посылаем каждому клиенту это сообщение
case text := <-pool.broadcasts:
for _, client := range pool.clients {
client.MsgChan <- text
}
}

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

Александр Павлюк: спасибо. Кстати у тебя баг был. При обрыве связи постоянно писало пустое сообщение. добавил проверку на EOF и теперь сижу как барун и не могу понять как удалить подключенного клиента

reader := bufio.NewReader(conn)
ILOOP:
	for {
		text, err := reader.ReadString('\n')
		switch err {
		case io.EOF:
			pool.DeleteClient(client)
			log.Printf("Перезапуск цикла")
			break ILOOP
		case nil:
			incomingClientMsg <- text
		default:
			log.Printf("Ошибка при получении:%s", err)
			//return
		}

	}

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

Владимир Грабко: Удалить клиента можно через pool.DeleteClient()
Поправить ошибку можно вот так, например: pastebin.com/WCMFB4di
Не обязательно проверять на EOF, достаточно просто смотреть, есть ли ошибка.

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

Александр Павлюк: спасибо!