| 82 | } |
| 83 | |
| 84 | func (s *FSSReconcileServer) Reconcile(_ *proto.ConnectRequest, stream proto.FSReconcile_ReconcileServer) error { |
| 85 | s.connected.Do(func() { |
| 86 | close(s.readyCh) |
| 87 | }) |
| 88 | if Status(atomic.LoadInt32(&s.status)) == Ready { |
| 89 | return fmt.Errorf("there is already a reconcile stream") |
| 90 | } |
| 91 | s.log.InfoContext(s.ctx, "reconcile client connected") |
| 92 | defer func() { |
| 93 | atomic.StoreInt32(&s.status, int32(NotReady)) |
| 94 | s.log.InfoContext(s.ctx, "grpc reconcile stream closed") |
| 95 | }() |
| 96 | // Sync all exiting function status to the newly connected reconcile client |
| 97 | s.functionsMu.Lock() |
| 98 | statusList := make([]*proto.FunctionStatus, 0, len(s.functions)) |
| 99 | for _, v := range s.functions { |
| 100 | statusList = append(statusList, v.status) |
| 101 | } |
| 102 | s.functionsMu.Unlock() |
| 103 | for _, v := range statusList { |
| 104 | err := stream.Send(v) |
| 105 | if err != nil { |
| 106 | s.log.ErrorContext(stream.Context(), "failed to send status update", slog.Any("status", v)) |
| 107 | // Continue to send the next status update. |
| 108 | } |
| 109 | } |
| 110 | for { |
| 111 | select { |
| 112 | case status := <-s.reconcile: |
| 113 | s.log.DebugContext(s.ctx, "sending status update", slog.Any("status", status)) |
| 114 | err := stream.Send(status) |
| 115 | if err != nil { |
| 116 | s.log.ErrorContext(stream.Context(), "failed to send status update", slog.Any("status", status)) |
| 117 | // Continue to send the next status update. |
| 118 | } |
| 119 | case <-stream.Context().Done(): |
| 120 | return nil |
| 121 | case <-s.ctx.Done(): |
| 122 | return nil |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | } |
| 127 | |
| 128 | func (s *FSSReconcileServer) UpdateStatus(_ context.Context, newStatus *proto.FunctionStatus) (*proto.Response, error) { |
| 129 | s.log.DebugContext(s.ctx, "received status update", slog.Any("status", newStatus)) |