Method Run

input/tcp.go:73–118

(inch chan<- *baker.Data)

}

func (s *TCP) Run(inch chan<- *baker.Data) error {
	s.setOutputChannel(inch)

	ctxLog := log.WithFields(log.Fields{"f": "Run"})

	addr, err := net.ResolveTCPAddr("tcp", s.Cfg.Listener)
	if err != nil {
		ctxLog.WithFields(log.Fields{"listener": s.Cfg.Listener}).Error("Can't resolve")
		return err
	}

	l, err := net.ListenTCP("tcp", addr)
	if err != nil {
		return err
	}
	defer l.Close()

	wg := sync.WaitGroup{}
	for atomic.LoadInt64(&s.stop) == 0 {
		l.SetDeadline(time.Now().Add(1 * time.Second))
		conn, err := l.AcceptTCP()
		if err != nil {
			if errors.Is(err, os.ErrDeadlineExceeded) {
				continue
			}
			ctxLog.WithFields(log.Fields{"error": err}).Error("Error while accepting")
		}

		ctxLog.WithFields(log.Fields{"addr": conn.RemoteAddr()}).Info("Connected")

		wg.Add(1)
		go func(conn *net.TCPConn) {
			defer func() {
				conn.Close()
				wg.Done()
			}()

			if err := s.handleStream(conn); err != nil {
				ctxLog.WithError(err).WithFields(log.Fields{"error": err}).Error("Error when handling stream")
			}
		}(conn)
	}

	wg.Wait()
	return nil
}

func (s *TCP) setOutputChannel(data chan<- *baker.Data) {
	s.data = data