scheduler

README

scheduler

Package scheduler contains a base set of schedulers to schedule update events.

See the main repository README.md for build and documentation details.

Documentation

Overview

Package scheduler implements a scheduler for Tarantool 3.0.

Index

• type EtcdWatch
  • func NewEtcdWatch(watcher clientv3.Watcher, prefix string) *EtcdWatch
  • func (s *EtcdWatch) Stop()
  • func (s *EtcdWatch) Wait(ctx context.Context) error
• type Periodic
  • func NewPeriodic(period time.Duration) *Periodic
  • func (s *Periodic) Stop()
  • func (s *Periodic) Wait(ctx context.Context) error

Examples

• EtcdWatch.Stop
• EtcdWatch.Wait
• EtcdWatch.Wait (ContextCancel)
• Periodic.Stop
• Periodic.Wait
• Periodic.Wait (ContextCancel)

Types

type EtcdWatch

type EtcdWatch struct {
	// contains filtered or unexported fields
}

EtcdWatch generates an update event when the configuration is updated in etcd. Configuration is searched by the prefix.

Note that the package user configures the connection to etcd himself.

func NewEtcdWatch

func NewEtcdWatch(watcher clientv3.Watcher,
	prefix string) *EtcdWatch

NewEtcdWatch creates an EtcdWatch scheduler with given watcher for the given prefix.

watcher is a watcher for etcd events from the etcd package. For example, clientv3.Client implements a clientv3.Watcher interface.

func (*EtcdWatch) Stop

func (s *EtcdWatch) Stop()

Stop stops the watcher and event generation.

Example

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery/scheduler"
)

func init() {
	log.SetOutput(io.Discard)
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	scheduler := scheduler.NewEtcdWatch(etcd, "key")

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

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		err := scheduler.Wait(ctx)
		if err != nil {
			fmt.Println("Error:", err.Error()+".")
		} else {
			fmt.Println("Event received.")
		}
		wg.Done()
	}()

	scheduler.Stop()
	wg.Wait()

}

Output:
Error: scheduler was stopped.

func (*EtcdWatch) Wait

func (s *EtcdWatch) Wait(ctx context.Context) error

Wait allows waiting until the configuration in etcd is updated.

Example

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery/scheduler"
)

func init() {
	log.SetOutput(io.Discard)
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	scheduler := scheduler.NewEtcdWatch(etcd, "key")

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

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		err := scheduler.Wait(ctx)
		if err != nil {
			fmt.Println("Error:", err.Error()+".")
		} else {
			fmt.Println("Event received.")
		}
		wg.Done()
	}()

	ctxPut, cancelPut := context.WithTimeout(context.Background(), 5*time.Second)
	_, err = etcd.Put(ctxPut, "key_prefix", "value")
	cancelPut()

	if err != nil {
		fmt.Println("Etcd put failed:", err)
		return
	}

	wg.Wait()

}

Output:
Event received.

Example (ContextCancel)

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"

	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/tests/v3/integration"

	"github.com/tarantool/go-discovery/scheduler"
)

func init() {
	log.SetOutput(io.Discard)
}

func main() {
	cluster := integration.NewLazyCluster()
	defer cluster.Terminate()

	etcd, err := clientv3.New(clientv3.Config{
		Endpoints: cluster.EndpointsGRPC(),
	})
	if err != nil {
		fmt.Println("Unable to start etcd client:", err)
		return
	}
	defer func() { _ = etcd.Close() }()

	scheduler := scheduler.NewEtcdWatch(etcd, "key")

	ctx, cancel := context.WithCancel(context.Background())

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		err := scheduler.Wait(ctx)
		if err != nil {
			fmt.Println("Error:", err.Error()+".")
		} else {
			fmt.Println("Event received.")
		}
		wg.Done()
	}()

	cancel()
	wg.Wait()

}

Output:
Error: context canceled.

type Periodic

type Periodic struct {
	// contains filtered or unexported fields
}

Periodic allows updating the configuration by timer, after each time period.

func NewPeriodic

func NewPeriodic(period time.Duration) *Periodic

NewPeriodic creates a Periodic with a given tick time period.

func (*Periodic) Stop

func (s *Periodic) Stop()

Stop makes Periodic satisfy the Scheduler interface.

Example

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/tarantool/go-discovery/scheduler"
)

func main() {
	scheduler := scheduler.NewPeriodic(10 * time.Second)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		err := scheduler.Wait(ctx)
		if err != nil {
			fmt.Println("Error:", err.Error()+".")
		} else {
			fmt.Println("Event tick.")
		}
		wg.Done()
	}()

	scheduler.Stop()
	wg.Wait()

}

Output:
Error: scheduler was stopped.

func (*Periodic) Wait

func (s *Periodic) Wait(ctx context.Context) error

Wait makes Periodic satisfy the Scheduler interface.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/tarantool/go-discovery/scheduler"
)

func main() {
	scheduler := scheduler.NewPeriodic(time.Second)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	err := scheduler.Wait(ctx)
	if err != nil {
		fmt.Println("Error:", err.Error()+".")
	} else {
		fmt.Println("Event tick.")
	}

}

Output:
Event tick.

Example (ContextCancel)

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/tarantool/go-discovery/scheduler"
)

func main() {
	scheduler := scheduler.NewPeriodic(10 * time.Second)
	ctx, cancel := context.WithCancel(context.Background())

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		err := scheduler.Wait(ctx)
		if err != nil {
			fmt.Println("Error:", err.Error()+".")
		} else {
			fmt.Println("Event tick.")
		}
		wg.Done()
	}()

	cancel()
	wg.Wait()

}

Output:
Error: context canceled.