Method GetOperationEvents

internal/api/backresthandler.go:364–442

GetOperationEvents implements GET /v1/events/operations

(ctx context.Context, req *connect.Request[emptypb.Empty], resp *connect.ServerStream[v1.OperationEvent])

// GetOperationEvents implements GET /v1/events/operations
func (s *BackrestHandler) GetOperationEvents(ctx context.Context, req *connect.Request[emptypb.Empty], resp *connect.ServerStream[v1.OperationEvent]) error {
	errChan := make(chan error, 1)
	events := make(chan *v1.OperationEvent, 100)

	timer := time.NewTicker(60 * time.Second)
	defer timer.Stop()

	callback := func(ops []*v1.Operation, eventType oplog.OperationEvent) {
		var event *v1.OperationEvent
		switch eventType {
		case oplog.OPERATION_ADDED:
			event = &v1.OperationEvent{
				Event: &v1.OperationEvent_CreatedOperations{
					CreatedOperations: &v1.OperationList{
						Operations: ops,
					},
				},
			}
		case oplog.OPERATION_UPDATED:
			event = &v1.OperationEvent{
				Event: &v1.OperationEvent_UpdatedOperations{
					UpdatedOperations: &v1.OperationList{
						Operations: ops,
					},
				},
			}
		case oplog.OPERATION_DELETED:
			ids := make([]int64, len(ops))
			for i, o := range ops {
				ids[i] = o.Id
			}

			event = &v1.OperationEvent{
				Event: &v1.OperationEvent_DeletedOperations{
					DeletedOperations: &types.Int64List{
						Values: ids,
					},
				},
			}
		default:
			zap.L().Error("Unknown event type")
		}

		select {
		case events <- event:
		default:
			select {
			case errChan <- errors.New("event buffer overflow, closing stream for client retry and catchup"):
			default:
			}
		}
	}

	s.oplog.Subscribe(oplog.SelectAll, &callback)
	defer func() {
		if err := s.oplog.Unsubscribe(&callback); err != nil {
			zap.L().Error("failed to unsubscribe from oplog", zap.Error(err))
		}