rake

Documentation

Index

• type Any
• type Rake
  • func New(rake func(a item), attr func(a item) interface{}, somany int) (my *Rake)
  • func (my *Rake) Attr(attr func(a item) interface{}) *Rake
  • func (my *Rake) Done() (done <-chan struct{})
  • func (my *Rake) Feed(items ...item) *Rake
  • func (my *Rake) Rake(rake func(a item)) *Rake

Examples

• Rake (Assign_rake)
• Rake (Chained)
• Rake (Closure)
• Rake (Closure_as_arg_dangerous)
• Rake (Closure_as_arg_stepbystep)
• Rake (Closure_on_embedded)

Types

type Any

type Any generic.Type

Any is the generic type flowing thru the pipe network.

type Rake

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

Rake represents a fanned out circular pipe network with a flexibly adjusting buffer. Any item is processed once only - items seen before are filtered out.

A Rake may be used e.g. as a crawling Crawler where every link shall be visited only once.

Example (Assign_rake)

c := &crawler{}
// c.Rake(New(c.put, nil, 80).Feed(1, 2, 3, 4, 5)) // this will bounce! nil!
c.Rake(New(c.put, nil, 80))
c.rake.Feed(1, 2, 3, 4, 5) // now everybody is in his place
<-c.rake.Done()

fmt.Println("Done")

Output:

Done

Example (Chained)

var r *Rake
crawl := func(item Any) {
	if false {
		log.Println("have:", item)
	}
	for i := 0; i < rand.Intn(9)+2; i++ {
		r.Feed(rand.Intn(2000)) // up to 10 new numbers < 2.000
	}
	time.Sleep(time.Millisecond)
}

r = New(nil, nil, 80).Rake(crawl).Feed(1) // chain

<-r.Done()

fmt.Println("Done")

Output:

Done

Example (Closure)

var r *Rake
crawl := func(item Any) {
	if false {
		log.Println("have:", item)
	}
	for i := 0; i < rand.Intn(9)+2; i++ {
		r.Feed(rand.Intn(2000)) // up to 10 new numbers < 2.000
	}
	time.Sleep(time.Millisecond)
}

r = New(crawl, nil, 80)

r.Feed(1)

<-r.Done()

fmt.Println("Done")

Output:

Done

Example (Closure_as_arg_dangerous)

// Very dangerous!
// works only because
// it takes longer for `1` to progress in the network
// than it takes for go to adjust the pointer `c`
// and thus the closure will have it when invoked.
var c *crawling                                               // need to declare `c` first so we may define the closure with feed back to `c`
c = &crawling{New(func(a Any) { c.put(a) }, nil, 80).Feed(1)} // pass the closure directly as func
<-c.Done()

fmt.Println("Done")

Output:

Done

Example (Closure_as_arg_stepbystep)

var c *crawling                                       // need to declare `c` first so we may define the closure with feed back to `c`
c = &crawling{New(func(a Any) { c.put(a) }, nil, 80)} // pass the closure directly as func
c.Feed(1)
<-c.Done()

fmt.Println("Done")

Output:

Done

Example (Closure_on_embedded)

var c *crawling // need to declare first so we may define the function with feed back

crawl := func(item Any) { c.put(item) } // a closure

c = &crawling{ // instantiate c
	New(crawl, nil, 80).Feed(1), // embed *Rake using closure `crawl`
}

<-c.Done()

fmt.Println("Done")

Output:

Done

func New

func New(
	rake func(a item),
	attr func(a item) interface{},
	somany int,
) (
	my *Rake,
)

New returns a (pointer to a) new operational Rake.

`rake` is the operation to be executed in parallel on any item which has not been seen before. Have it use `myrake.Feed(items...)` in order to provide feed-back.

`attr` allows to specify an attribute for the seen filter. Pass `nil` to filter on any item itself.

`somany` is the # of parallel processes - the parallelism of the network built by Rake, the # of parallel raking endpoints of the Rake.

func (*Rake) Attr

func (my *Rake) Attr(attr func(a item) interface{}) *Rake

Attr sets the (optional) attribute to discriminate 'seen'.

`attr` allows to specify an attribute for the 'seen' filter. If not set 'seen' will discriminate any item by itself.

Seen panics iff called after first nonempty `Feed(...)`

func (*Rake) Done

func (my *Rake) Done() (done <-chan struct{})

Done returns a channel which will be signalled and closed when traffic has subsided, nothing is left to be processed and consequently all goroutines have terminated.

func (*Rake) Feed

func (my *Rake) Feed(items ...item) *Rake

Feed registers new items on the network.

func (*Rake) Rake

func (my *Rake) Rake(rake func(a item)) *Rake

Rake sets the rake function to be applied (in parallel).

`rake` is the operation to be executed in parallel on any item which has not been seen before.

You may provide `nil` here and call `Rake(..)` later to provide it. Or have it use `myrake.Feed(items...)` in order to provide feed-back.

Rake panics iff called after first nonempty `Feed(...)`