Changeset 210 in code for trunk/downstream.go


Ignore:
Timestamp:
Apr 3, 2020, 2:35:08 PM (5 years ago)
Author:
contact
Message:

Introduce conn for common connection logic

This centralizes the common upstream & downstream bits.

File:
1 edited

Legend:

Unmodified
Added
Removed

  • trunk/downstream.go

    r209 r210  
    5353
    5454type downstreamConn struct {
    55         id       uint64
    56         net      net.Conn
    57         irc      *irc.Conn
    58         srv      *Server
    59         logger   Logger
    60         outgoing chan<- *irc.Message
    61         closed   chan struct{}
     55        conn
     56
     57        id uint64
    6258
    6359        registered  bool
    … …  
    8581
    8682func newDownstreamConn(srv *Server, netConn net.Conn, id uint64) *downstreamConn {
    87         outgoing := make(chan *irc.Message, 64)
     83        logger := &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())}
    8884        dc := &downstreamConn{
     85                conn:          *newConn(srv, netConn, logger),
    8986                id:            id,
    90                 net:           netConn,
    91                 irc:           irc.NewConn(netConn),
    92                 srv:           srv,
    93                 logger:        &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
    94                 outgoing:      outgoing,
    95                 closed:        make(chan struct{}),
    9687                ringConsumers: make(map[*network]*RingConsumer),
    9788                caps:          make(map[string]bool),
    … …  
    10293                dc.hostname = host
    10394        }
    104 
    105         go func() {
    106                 for msg := range outgoing {
    107                         if dc.srv.Debug {
    108                                 dc.logger.Printf("sent: %v", msg)
    109                         }
    110                         dc.net.SetWriteDeadline(time.Now().Add(writeTimeout))
    111                         if err := dc.irc.WriteMessage(msg); err != nil {
    112                                 dc.logger.Printf("failed to write message: %v", err)
    113                                 break
    114                         }
    115                 }
    116                 if err := dc.net.Close(); err != nil {
    117                         dc.logger.Printf("failed to close connection: %v", err)
    118                 } else {
    119                         dc.logger.Printf("connection closed")
    120                 }
    121                 // Drain the outgoing channel to prevent SendMessage from blocking
    122                 for range outgoing {
    123                         // This space is intentionally left blank
    124                 }
    125         }()
    126 
    127         dc.logger.Printf("new connection")
    12895        return dc
    12996}
    … …  
    228195}
    229196
    230 func (dc *downstreamConn) isClosed() bool {
    231         select {
    232         case <-dc.closed:
    233                 return true
    234         default:
    235                 return false
    236         }
    237 }
    238 
    239197func (dc *downstreamConn) readMessages(ch chan<- event) error {
    240198        for {
    241                 msg, err := dc.irc.ReadMessage()
     199                msg, err := dc.ReadMessage()
    242200                if err == io.EOF {
    243201                        break
    … …  
    246204                }
    247205
    248                 if dc.srv.Debug {
    249                         dc.logger.Printf("received: %v", msg)
    250                 }
    251 
    252206                ch <- eventDownstreamMessage{msg, dc}
    253207        }
    … …  
    256210}
    257211
    258 func (dc *downstreamConn) writeMessages() error {
    259         return nil
    260 }
    261 
    262 // Close closes the connection. It is safe to call from any goroutine.
    263 func (dc *downstreamConn) Close() error {
    264         if dc.isClosed() {
    265                 return fmt.Errorf("downstream connection already closed")
    266         }
    267         close(dc.closed)
    268         close(dc.outgoing)
    269         return nil
    270 }
    271 
    272 // SendMessage queues a new outgoing message. It is safe to call from any
    273 // goroutine.
    274212func (dc *downstreamConn) SendMessage(msg *irc.Message) {
    275         if dc.isClosed() {
    276                 return
    277         }
    278213        // TODO: strip tags if the client doesn't support them (see runNetwork)
    279         dc.outgoing <- msg
     214        dc.conn.SendMessage(msg)
    280215}
    281216
Note: See TracChangeset for help on using the changeset viewer.