dataloader

Documentation

Overview

Example

package main

import (
	"context"
	"fmt"
	"sync/atomic"

	"fillmore-labs.com/microbatch"
	"fillmore-labs.com/microbatch/dataloader"
	"fillmore-labs.com/promise"
)

type DataProcessor struct {
	Calls atomic.Int32
	Keys  atomic.Int32
}

type QueryResult struct {
	ID    int
	Value string
}

func (p *DataProcessor) ProcessJobs(keys []int) ([]QueryResult, error) {
	p.Calls.Add(1)
	p.Keys.Add(int32(len(keys)))
	results := make([]QueryResult, 0, len(keys))
	for _, key := range keys {
		result := QueryResult{
			ID:    key,
			Value: fmt.Sprintf("Query Result %d", key),
		}
		results = append(results, result)
	}

	return results, nil
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	p := &DataProcessor{}
	d := dataloader.NewDataLoader(
		p.ProcessJobs,
		func(q QueryResult) int { return q.ID },
		microbatch.WithSize(3),
	)

	queries := [11]int{1, 2, 1, 2, 3, 3, 4, 1, 2, 3, 5}
	results := make([]*promise.Memoizer[QueryResult], len(queries))
	for i, query := range queries {
		results[i] = d.Load(query)
	}

	d.Send()

	// Wait for all queries to complete
	var err error
	for _, result := range results {
		if _, e := result.Await(ctx); e != nil {
			err = e
		}
	}

	if err == nil {
		fmt.Printf("Requested %d keys in %d calls\n", p.Keys.Load(), p.Calls.Load())
	}

}

Output:

Requested 5 keys in 2 calls

Index

• type DataLoader
  • func NewDataLoader[K comparable, R any, KK ~[]K, RR ~[]R](processJobs func(keys KK) (RR, error), correlate func(result R) K, ...) *DataLoader[K, R]
  • func (d *DataLoader[K, R]) Load(key K) *promise.Memoizer[R]
  • func (d *DataLoader[K, R]) Send()

Examples

• Package

Types

type DataLoader

type DataLoader[K comparable, R any] struct {
	// contains filtered or unexported fields
}

DataLoader demonstrates how to use microbatch.Batcher to implement a simple Facebook DataLoader. K and R define the key and result types for batching.

func NewDataLoader

func NewDataLoader[K comparable, R any, KK ~[]K, RR ~[]R](
	processJobs func(keys KK) (RR, error),
	correlate func(result R) K,
	opts ...microbatch.Option,
) *DataLoader[K, R]

NewDataLoader create a new DataLoader.

func (*DataLoader[K, R]) Load

func (d *DataLoader[K, R]) Load(key K) *promise.Memoizer[R]

Load retrieves a value from the cache or loads it asynchronously.

func (*DataLoader[K, R]) Send

func (d *DataLoader[K, R]) Send()

Send loads all submitted keys.