@Summary Workspace Agent Connection Watch @ID workspace-agent-connection-watch @Security CoderSessionToken @Produce json @Tags Workspaces @Param workspace path string true "Workspace ID" format(uuid) @Success 101 {object} workspacesdk.ConnectionWatchEvent @Router /api/v2/workspaces/{workspace}/agent
(rw http.ResponseWriter, r *http.Request)
| 64 | // @Success 101 {object} workspacesdk.ConnectionWatchEvent |
| 65 | // @Router /api/v2/workspaces/{workspace}/agent-connection-watch [get] |
| 66 | func (w *Watcher) WorkspaceAgentConnectionWatch(rw http.ResponseWriter, r *http.Request) { |
| 67 | ctx := r.Context() |
| 68 | workspace := httpmw.WorkspaceParam(r) |
| 69 | agentName := r.URL.Query().Get("agent_name") |
| 70 | |
| 71 | filteredEvents := make(chan event, 1) |
| 72 | filteredEvents <- event{sync: true} // init sync |
| 73 | cancelWorkspaceSubscribe, err := w.sub.SubscribeWithErr(wspubsub.WorkspaceEventChannel(workspace.OwnerID), |
| 74 | wspubsub.HandleWorkspaceEvent( |
| 75 | func(ctx context.Context, payload wspubsub.WorkspaceEvent, err error) { |
| 76 | if err != nil { |
| 77 | // subscription error, resync |
| 78 | select { |
| 79 | case filteredEvents <- event{sync: true}: |
| 80 | case <-ctx.Done(): |
| 81 | } |
| 82 | return |
| 83 | } |
| 84 | if payload.WorkspaceID != workspace.ID { |
| 85 | return |
| 86 | } |
| 87 | select { |
| 88 | case filteredEvents <- event{wsEvent: &payload}: |
| 89 | case <-ctx.Done(): |
| 90 | } |
| 91 | })) |
| 92 | if err != nil { |
| 93 | w.logger.Error(ctx, "failed to subscribe to workspace events", |
| 94 | slog.Error(err), slog.F("owner_id", workspace.OwnerID)) |
| 95 | httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{ |
| 96 | Message: "Internal error setting up workspace event subscription", |
| 97 | // Don't include the error in case it leaks infra details about the pubsub |
| 98 | }) |
| 99 | return |
| 100 | } |
| 101 | defer cancelWorkspaceSubscribe() |
| 102 | |
| 103 | closed := false |
| 104 | w.mu.Lock() |
| 105 | closed = w.closed |
| 106 | if !closed { |
| 107 | w.wg.Add(1) |
| 108 | } |
| 109 | w.mu.Unlock() |
| 110 | if closed { |
| 111 | w.logger.Debug(ctx, "server is closed, writing error") |
| 112 | httpapi.Write(ctx, rw, http.StatusServiceUnavailable, codersdk.Response{ |
| 113 | Message: "Server instance is shutting down", |
| 114 | }) |
| 115 | return |
| 116 | } |
| 117 | defer w.wg.Done() |
| 118 | |
| 119 | conn, err := websocket.Accept(rw, r, nil) |
| 120 | if err != nil { |
| 121 | httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{ |
| 122 | Message: "Failed to accept WebSocket.", |
| 123 | Detail: err.Error(), |
nothing calls this directly
no test coverage detected