|
17 | 17 | package desktop |
18 | 18 |
|
19 | 19 | import ( |
| 20 | + "bytes" |
20 | 21 | "context" |
21 | 22 | "encoding/json" |
| 23 | + "errors" |
22 | 24 | "fmt" |
| 25 | + "io" |
23 | 26 | "net" |
24 | 27 | "net/http" |
25 | 28 | "strings" |
26 | 29 |
|
27 | 30 | "github.com/docker/compose/v2/internal/memnet" |
| 31 | + "github.com/r3labs/sse" |
28 | 32 | "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" |
29 | 33 | ) |
30 | 34 |
|
@@ -119,6 +123,175 @@ func (c *Client) FeatureFlags(ctx context.Context) (FeatureFlagResponse, error) |
119 | 123 | return ret, nil |
120 | 124 | } |
121 | 125 |
|
| 126 | +type CreateFileShareRequest struct { |
| 127 | + HostPath string `json:"hostPath"` |
| 128 | + Labels map[string]string `json:"labels,omitempty"` |
| 129 | +} |
| 130 | + |
| 131 | +type CreateFileShareResponse struct { |
| 132 | + FileShareID string `json:"fileShareID"` |
| 133 | +} |
| 134 | + |
| 135 | +func (c *Client) CreateFileShare(ctx context.Context, r CreateFileShareRequest) (*CreateFileShareResponse, error) { |
| 136 | + rawBody, _ := json.Marshal(r) |
| 137 | + req, err := http.NewRequestWithContext(ctx, http.MethodPost, backendURL("/mutagen/file-shares"), bytes.NewReader(rawBody)) |
| 138 | + req.Header.Set("Content-Type", "application/json") |
| 139 | + if err != nil { |
| 140 | + return nil, err |
| 141 | + } |
| 142 | + resp, err := c.client.Do(req) |
| 143 | + if err != nil { |
| 144 | + return nil, err |
| 145 | + } |
| 146 | + defer func() { |
| 147 | + _ = resp.Body.Close() |
| 148 | + }() |
| 149 | + |
| 150 | + if resp.StatusCode != http.StatusOK { |
| 151 | + errBody, _ := io.ReadAll(resp.Body) |
| 152 | + return nil, fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, string(errBody)) |
| 153 | + } |
| 154 | + var ret CreateFileShareResponse |
| 155 | + if err := json.NewDecoder(resp.Body).Decode(&ret); err != nil { |
| 156 | + return nil, err |
| 157 | + } |
| 158 | + return &ret, nil |
| 159 | +} |
| 160 | + |
| 161 | +type FileShareReceiverState struct { |
| 162 | + TotalReceivedSize uint64 `json:"totalReceivedSize"` |
| 163 | +} |
| 164 | + |
| 165 | +type FileShareEndpoint struct { |
| 166 | + Path string `json:"path"` |
| 167 | + TotalFileSize uint64 `json:"totalFileSize,omitempty"` |
| 168 | + StagingProgress *FileShareReceiverState `json:"stagingProgress"` |
| 169 | +} |
| 170 | + |
| 171 | +type FileShareSession struct { |
| 172 | + SessionID string `json:"identifier"` |
| 173 | + Alpha FileShareEndpoint `json:"alpha"` |
| 174 | + Beta FileShareEndpoint `json:"beta"` |
| 175 | + Labels map[string]string `json:"labels"` |
| 176 | + Status string `json:"status"` |
| 177 | +} |
| 178 | + |
| 179 | +func (c *Client) ListFileShares(ctx context.Context) ([]FileShareSession, error) { |
| 180 | + req, err := http.NewRequestWithContext(ctx, http.MethodGet, backendURL("/mutagen/file-shares"), http.NoBody) |
| 181 | + if err != nil { |
| 182 | + return nil, err |
| 183 | + } |
| 184 | + resp, err := c.client.Do(req) |
| 185 | + if err != nil { |
| 186 | + return nil, err |
| 187 | + } |
| 188 | + defer func() { |
| 189 | + _ = resp.Body.Close() |
| 190 | + }() |
| 191 | + |
| 192 | + if resp.StatusCode != http.StatusOK { |
| 193 | + return nil, newHTTPStatusCodeError(resp) |
| 194 | + } |
| 195 | + |
| 196 | + var ret []FileShareSession |
| 197 | + if err := json.NewDecoder(resp.Body).Decode(&ret); err != nil { |
| 198 | + return nil, err |
| 199 | + } |
| 200 | + return ret, nil |
| 201 | +} |
| 202 | + |
| 203 | +func (c *Client) DeleteFileShare(ctx context.Context, id string) error { |
| 204 | + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, backendURL("/mutagen/file-shares/"+id), http.NoBody) |
| 205 | + if err != nil { |
| 206 | + return err |
| 207 | + } |
| 208 | + resp, err := c.client.Do(req) |
| 209 | + if err != nil { |
| 210 | + return err |
| 211 | + } |
| 212 | + defer func() { |
| 213 | + _ = resp.Body.Close() |
| 214 | + }() |
| 215 | + if resp.StatusCode < 200 || resp.StatusCode >= 300 { |
| 216 | + return newHTTPStatusCodeError(resp) |
| 217 | + } |
| 218 | + return nil |
| 219 | +} |
| 220 | + |
| 221 | +type EventMessage[T any] struct { |
| 222 | + Value T |
| 223 | + Error error |
| 224 | +} |
| 225 | + |
| 226 | +func newHTTPStatusCodeError(resp *http.Response) error { |
| 227 | + r := io.LimitReader(resp.Body, 2048) |
| 228 | + body, err := io.ReadAll(r) |
| 229 | + if err != nil { |
| 230 | + return fmt.Errorf("http status code %d", resp.StatusCode) |
| 231 | + } |
| 232 | + return fmt.Errorf("http status code %d: %s", resp.StatusCode, string(body)) |
| 233 | +} |
| 234 | + |
| 235 | +func (c *Client) StreamFileShares(ctx context.Context) (<-chan EventMessage[[]FileShareSession], error) { |
| 236 | + req, err := http.NewRequestWithContext(ctx, http.MethodGet, backendURL("/mutagen/file-shares/stream"), http.NoBody) |
| 237 | + if err != nil { |
| 238 | + return nil, err |
| 239 | + } |
| 240 | + resp, err := c.client.Do(req) |
| 241 | + if err != nil { |
| 242 | + return nil, err |
| 243 | + } |
| 244 | + |
| 245 | + if resp.StatusCode < 200 || resp.StatusCode >= 300 { |
| 246 | + defer func() { |
| 247 | + _ = resp.Body.Close() |
| 248 | + }() |
| 249 | + return nil, newHTTPStatusCodeError(resp) |
| 250 | + } |
| 251 | + |
| 252 | + events := make(chan EventMessage[[]FileShareSession]) |
| 253 | + go func(ctx context.Context) { |
| 254 | + defer func() { |
| 255 | + _ = resp.Body.Close() |
| 256 | + for range events { |
| 257 | + // drain the channel |
| 258 | + } |
| 259 | + close(events) |
| 260 | + }() |
| 261 | + if err := readEvents(ctx, resp.Body, events); err != nil { |
| 262 | + select { |
| 263 | + case <-ctx.Done(): |
| 264 | + case events <- EventMessage[[]FileShareSession]{Error: err}: |
| 265 | + } |
| 266 | + } |
| 267 | + }(ctx) |
| 268 | + return events, nil |
| 269 | +} |
| 270 | + |
| 271 | +func readEvents[T any](ctx context.Context, r io.Reader, events chan<- EventMessage[T]) error { |
| 272 | + eventReader := sse.NewEventStreamReader(r) |
| 273 | + for { |
| 274 | + msg, err := eventReader.ReadEvent() |
| 275 | + if errors.Is(err, io.EOF) { |
| 276 | + return nil |
| 277 | + } else if err != nil { |
| 278 | + return fmt.Errorf("reading events: %w", err) |
| 279 | + } |
| 280 | + msg = bytes.TrimPrefix(msg, []byte("data: ")) |
| 281 | + |
| 282 | + var event T |
| 283 | + if err := json.Unmarshal(msg, &event); err != nil { |
| 284 | + return err |
| 285 | + } |
| 286 | + select { |
| 287 | + case <-ctx.Done(): |
| 288 | + return context.Cause(ctx) |
| 289 | + case events <- EventMessage[T]{Value: event}: |
| 290 | + // event was sent to channel, read next |
| 291 | + } |
| 292 | + } |
| 293 | +} |
| 294 | + |
122 | 295 | // backendURL generates a URL for the given API path. |
123 | 296 | // |
124 | 297 | // NOTE: Custom transport handles communication. The host is to create a valid |
|
0 commit comments