| 737 | } |
| 738 | |
| 739 | func (js *js) resetPendingAcksOnReconnect() { |
| 740 | js.mu.Lock() |
| 741 | connStatusCh := js.connStatusCh |
| 742 | js.mu.Unlock() |
| 743 | for { |
| 744 | newStatus, ok := <-connStatusCh |
| 745 | if !ok || newStatus == CLOSED { |
| 746 | return |
| 747 | } |
| 748 | js.mu.Lock() |
| 749 | errCb := js.opts.aecb |
| 750 | for id, paf := range js.pafs { |
| 751 | paf.err = ErrDisconnected |
| 752 | if paf.errCh != nil { |
| 753 | paf.errCh <- paf.err |
| 754 | } |
| 755 | if errCb != nil { |
| 756 | defer errCb(js, paf.msg, ErrDisconnected) |
| 757 | } |
| 758 | delete(js.pafs, id) |
| 759 | } |
| 760 | if js.dch != nil { |
| 761 | close(js.dch) |
| 762 | js.dch = nil |
| 763 | } |
| 764 | js.mu.Unlock() |
| 765 | } |
| 766 | } |
| 767 | |
| 768 | // CleanupPublisher will cleanup the publishing side of JetStreamContext. |
| 769 | // |