Method Stream

app/store/database/pullreq.go:514–559

Stream returns a list of pull requests for a repo.

(ctx context.Context, opts *types.PullReqFilter)

// Stream returns a list of pull requests for a repo.
func (s *PullReqStore) Stream(ctx context.Context, opts *types.PullReqFilter) (<-chan *types.PullReq, <-chan error) {
	stmt := s.listQuery(opts)

	stmt = stmt.OrderBy("pullreq_updated desc")

	chPRs := make(chan *types.PullReq)
	chErr := make(chan error, 1)

	go func() {
		defer close(chPRs)
		defer close(chErr)

		sql, args, err := stmt.ToSql()
		if err != nil {
			chErr <- fmt.Errorf("failed to convert query to sql: %w", err)
			return
		}

		db := dbtx.GetAccessor(ctx, s.db)

		rows, err := db.QueryxContext(ctx, sql, args...)
		if err != nil {
			chErr <- database.ProcessSQLErrorf(ctx, err, "Failed to execute stream query")
			return
		}

		defer func() { _ = rows.Close() }()

		for rows.Next() {
			var prData pullReq
			err = rows.StructScan(&prData)
			if err != nil {
				chErr <- fmt.Errorf("failed to scan pull request: %w", err)
				return
			}

			chPRs <- s.mapPullReq(ctx, &prData)
		}

		if err := rows.Err(); err != nil {
			chErr <- fmt.Errorf("failed to scan pull request: %w", err)
		}
	}()

	return chPRs, chErr
}

func (s *PullReqStore) ListOpenByBranchName(
	ctx context.Context,