| 552 | } |
| 553 | |
| 554 | func (c *connection) waitFlush() (err error) { |
| 555 | timeout := c.writeTimeout |
| 556 | if dl := c.writeDeadline; dl > 0 { |
| 557 | timeout = time.Duration(dl - time.Now().UnixNano()) |
| 558 | if timeout <= 0 { |
| 559 | return Exception(ErrWriteTimeout, c.remoteAddr.String()) |
| 560 | } |
| 561 | } |
| 562 | if timeout == 0 { |
| 563 | return <-c.writeTrigger |
| 564 | } |
| 565 | |
| 566 | // set write timeout |
| 567 | if c.writeTimer == nil { |
| 568 | c.writeTimer = time.NewTimer(timeout) |
| 569 | } else { |
| 570 | c.writeTimer.Reset(timeout) |
| 571 | } |
| 572 | |
| 573 | select { |
| 574 | case err = <-c.writeTrigger: |
| 575 | if !c.writeTimer.Stop() { // clean timer |
| 576 | <-c.writeTimer.C |
| 577 | } |
| 578 | return err |
| 579 | case <-c.writeTimer.C: |
| 580 | select { |
| 581 | // try fetch writeTrigger if both cases fires |
| 582 | case err = <-c.writeTrigger: |
| 583 | return err |
| 584 | default: |
| 585 | } |
| 586 | // if timeout, remove write event from poller |
| 587 | // we cannot flush it again, since we don't if the poller is still process outputBuffer |
| 588 | c.operator.Control(PollRW2R) |
| 589 | return Exception(ErrWriteTimeout, c.remoteAddr.String()) |
| 590 | } |
| 591 | } |
| 592 | |
| 593 | func (c *connection) getState() connState { |
| 594 | return atomic.LoadInt32(&c.state) |