| 139 | } |
| 140 | |
| 141 | func (c *coordinator) Coordinate( |
| 142 | ctx context.Context, id uuid.UUID, name string, a CoordinateeAuth, |
| 143 | ) ( |
| 144 | chan<- *proto.CoordinateRequest, <-chan *proto.CoordinateResponse, |
| 145 | ) { |
| 146 | logger := c.core.logger.With( |
| 147 | slog.F("peer_id", id), |
| 148 | slog.F("peer_name", name), |
| 149 | ) |
| 150 | reqs := make(chan *proto.CoordinateRequest, RequestBufferSize) |
| 151 | resps := make(chan *proto.CoordinateResponse, ResponseBufferSize) |
| 152 | |
| 153 | p := &peer{ |
| 154 | logger: logger, |
| 155 | id: id, |
| 156 | name: name, |
| 157 | resps: resps, |
| 158 | reqs: reqs, |
| 159 | auth: a, |
| 160 | sent: make(map[uuid.UUID]*proto.Node), |
| 161 | } |
| 162 | err := c.core.initPeer(p) |
| 163 | if err != nil { |
| 164 | if xerrors.Is(err, ErrClosed) { |
| 165 | logger.Debug(ctx, "coordinate failed: Coordinator is closed") |
| 166 | } else { |
| 167 | logger.Critical(ctx, "coordinate failed", slog.Error(err)) |
| 168 | } |
| 169 | } |
| 170 | c.mu.Lock() |
| 171 | defer c.mu.Unlock() |
| 172 | if c.closed { |
| 173 | // don't start the readLoop if we are closed. |
| 174 | return reqs, resps |
| 175 | } |
| 176 | c.wg.Add(1) |
| 177 | go func() { |
| 178 | defer c.wg.Done() |
| 179 | loopErr := p.reqLoop(ctx, logger, c.core.handleRequest) |
| 180 | closeErrStr := "" |
| 181 | if loopErr != nil { |
| 182 | closeErrStr = loopErr.Error() |
| 183 | } |
| 184 | err := c.core.lostPeer(p, closeErrStr) |
| 185 | if xerrors.Is(err, ErrClosed) || xerrors.Is(err, ErrAlreadyRemoved) { |
| 186 | return |
| 187 | } |
| 188 | if err != nil { |
| 189 | logger.Error(context.Background(), "failed to process lost peer", slog.Error(err)) |
| 190 | } |
| 191 | }() |
| 192 | return reqs, resps |
| 193 | } |
| 194 | |
| 195 | // core is an in-memory structure of peer mappings. Its methods may be called from multiple goroutines; |
| 196 | // it is protected by a mutex to ensure data stay consistent. |