()
| 742 | } |
| 743 | |
| 744 | func (m *mapper) run() { |
| 745 | for { |
| 746 | var best map[uuid.UUID]mapping |
| 747 | select { |
| 748 | case <-m.ctx.Done(): |
| 749 | return |
| 750 | case mappings := <-m.mappings: |
| 751 | m.logger.Debug(m.ctx, "got new mappings") |
| 752 | m.c.setLatestMapping(mappings) |
| 753 | best = m.bestMappings(mappings) |
| 754 | case <-m.update: |
| 755 | m.logger.Debug(m.ctx, "triggered update") |
| 756 | best = m.bestMappings(m.c.getLatestMapping()) |
| 757 | } |
| 758 | update := m.bestToUpdate(best) |
| 759 | if update == nil { |
| 760 | m.logger.Debug(m.ctx, "skipping nil node update") |
| 761 | continue |
| 762 | } |
| 763 | for _, chunk := range update.Chunked() { |
| 764 | if err := m.c.Enqueue(chunk); err != nil { |
| 765 | // lots of reasons this could happen, most usually, the peer has disconnected. |
| 766 | m.logger.Debug(m.ctx, "failed to enqueue chunk", slog.Error(err)) |
| 767 | break |
| 768 | } |
| 769 | } |
| 770 | } |
| 771 | } |
| 772 | |
| 773 | // bestMappings takes a set of mappings and resolves the best set of nodes. We may get several mappings for a |
| 774 | // particular connection, from different coordinators in the distributed system. Furthermore, some coordinators |
no test coverage detected