// transport will grab message arrival times, wait until that time, and
// then write the message out when it is scheduled to arrive
transport will grab message arrival times, wait until that time, and
then write the message out when it is scheduled to arrive
// transport will grab message arrival times wait until that time and // then write the message out when it is scheduled to arrive
func (s *stream) transport() {
	defer s.teardown()
	bufsize := 256
	buf := new(bytes.Buffer)
	timer := time.NewTimer(0)
	if !timer.Stop() {
		select {
		case <-timer.C:
		default:
		}
	}
	defer timer.Stop()
	drainBuf := func() error {
		if buf.Len() > 0 {
			_, err := s.write.Write(buf.Bytes())
			if err != nil {
				return err
			}
			buf.Reset()
		}
		return nil
	}
	deliverOrWait := func(o *transportObject) error {
		buffered := len(o.msg) + buf.Len()
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}
		delay := time.Until(o.arrivalTime)
		if delay >= 0 {
			timer.Reset(delay)
		} else {
			timer.Reset(0)
		}
		if buffered >= bufsize {
			select {
			case <-timer.C:
			case <-s.reset:
				select {
				case s.reset <- struct{}{}:
				default:
				}
				return mux.ErrReset
			}
			if err := drainBuf(); err != nil {
				return err
			}
			_, err := s.write.Write(o.msg)
			if err != nil {
				return err
			}
		} else {
			buf.Write(o.msg)
		}
		return nil
	}
	for {
		select {
		case <-s.reset:
			s.writeErr = mux.ErrReset
			return
		default:
		}
		select {
		case <-s.reset:
			s.writeErr = mux.ErrReset
			return
		case <-s.close:
			if err := drainBuf(); err != nil {
				s.cancelWrite(err)
				return
			}
			s.writeErr = s.write.Close()
			if s.writeErr == nil {
				s.writeErr = ErrClosed
			}
			return
		case o := <-s.toDeliver:
			if err := deliverOrWait(o); err != nil {
				s.cancelWrite(err)
				return
			}
		case <-timer.C: 
			if err := drainBuf(); err != nil {
				s.cancelWrite(err)
				return
			}
		}
	}
}
func ( s * stream ) transport ( ) { defer s . teardown ( ) bufsize := 256 buf := new ( bytes . Buffer ) timer := time . NewTimer ( 0 ) if ! timer . Stop ( ) { select { case <- timer . C : default : } } defer timer . Stop ( ) drainBuf := func ( ) error { if buf . Len ( ) > 0 { _ , err := s . write . Write ( buf . Bytes ( ) ) if err != nil { return err } buf . Reset ( ) } return nil } deliverOrWait := func ( o * transportObject ) error { buffered := len ( o . msg ) + buf . Len ( ) if ! timer . Stop ( ) { select { case <- timer . C : default : } } delay := time . Until ( o . arrivalTime ) if delay >= 0 { timer . Reset ( delay ) } else { timer . Reset ( 0 ) } if buffered >= bufsize { select { case <- timer . C : case <- s . reset : select { case s . reset <- struct { } { } : default : } return mux . ErrReset } if err := drainBuf ( ) ; err != nil { return err } _ , err := s . write . Write ( o . msg ) if err != nil { return err } } else { buf . Write ( o . msg ) } return nil } for { select { case <- s . reset : s . writeErr = mux . ErrReset return default : } select { case <- s . reset : s . writeErr = mux . ErrReset return case <- s . close : if err := drainBuf ( ) ; err != nil { s . cancelWrite ( err ) return } s . writeErr = s . write . Close ( ) if s . writeErr == nil { s . writeErr = ErrClosed } return case o := <- s . toDeliver : if err := deliverOrWait ( o ) ; err != nil { s . cancelWrite ( err ) return } case <- timer . C : if err := drainBuf ( ) ; err != nil { s . cancelWrite ( err ) return } } } }
transport will grab message arrival times, wait until that time, and
then write the message out when it is scheduled to arrive
transport will grab message arrival times wait until that time and then write the message out when it is scheduled to arrive
// cleanup // writeBuf writes the contents of buf through to the s.Writer. // done only when arrival time makes sense. // deliverOrWait is a helper func that processes // an incoming packet. it waits until the arrival time, // and then writes things out. // Yes, we can end up extending a timer multiple times if we // keep on making small writes but that shouldn't be too much of an // issue. Fixing that would be painful. // FIXME: So, we *shouldn't* need to do this but we hang // here if we don't... Go bug? // write this message. // Reset takes precedent. // ok, due to write it out.