Method Watch

internal/servers/watch_server.go:29–77

Watch function sets up a stream for the client to receive changes.

(request *v1.WatchRequest, server v1.Watch_WatchServer)

// Watch function sets up a stream for the client to receive changes.
func (r *WatchServer) Watch(request *v1.WatchRequest, server v1.Watch_WatchServer) error {
	// Start a new context and span for tracing.
	ctx, span := internal.Tracer.Start(server.Context(), "watch.watch")
	defer span.End() // Ensure the span ends when the function returns.

	// Validate the incoming request.
	v := request.Validate()
	if v != nil {
		return v // Return validation error, if any.
	}

	// Extract the snapshot token from the request.
	snap := request.GetSnapToken()
	if snap == "" {
		// If the snapshot token is not provided, get the head snapshot from the database.
		st, err := r.dr.HeadSnapshot(ctx, request.GetTenantId())
		if err != nil {
			return err // If there's an error retrieving the snapshot, return it.
		}
		// Encode the snapshot to a string.
		snap = st.Encode().String()
	}

	// Call the Watch function on the watcher, which returns two channels.
	changes, errs := r.w.Watch(ctx, request.GetTenantId(), snap)

	// Create a separate goroutine to handle sending changes to the server.
	go func() {
		for change := range changes {
			// For each change, send it to the client.
			if err := server.Send(&v1.WatchResponse{Changes: change}); err != nil {
				// If an error occurs while sending, exit the goroutine.
				// The error is not handled here because the context will be cancelled if an error is detected.
				return
			}
		}
	}()

	// Main loop for handling errors.
	for err := range errs {
		// If an error occurs, convert it to a status error and return it.
		// This ends the Watch function, which in turn closes the changes channel and ends the above goroutine.
		return status.Error(GetStatus(err), err.Error())
	}

	// At this point, the errs channel has been closed, indicating that no more errors will be coming in.
	// Therefore, it's safe to return nil indicating that the operation was successful.
	return nil
}