<commit_before>package subscription

import (
	"time"

	"github.com/jochasinga/grx/bases"
	"github.com/jochasinga/grx/subject"
)

// Subscription is usually returned from any subscription
type Subscription struct {
	observable    *subject.Subject
	SubscribeAt   time.Time
	UnsubscribeAt time.Time
}

// DefaultSubscription is a default Subscription
var DefaultSubscription = &Subscription{}

func New(fs ...func(*Subscription)) *Subscription {
	s := DefaultSubscription
	if len(fs) > 0 {
		for _, f := range fs {
			f(s)
		}
	}
	return s
}

// Dispose cleans up an observable and notify its unsubscribe channel and return a Subscriptor
func (s *Subscription) Dispose() bases.Subscriptor {
	go func() {
		s.observable.Unsubscribe()
		s.UnsubscribeAt = time.Now()
	}()
	return s
}

func (s *Subscription) Subscribe() bases.Subscriptor {
	s.SubscribeAt = time.Now()
	return s
}

// Unsubscribe is an alias for Dispose
func (s *Subscription) Unsubscribe() bases.Subscriptor {
	return s.Dispose()
}

// UnscribeIn notify the unsubscribe channel in d duration, then return the Subscriptor
func (s *Subscription) UnsubscribeIn(d time.Duration) <-chan bases.Subscriptor {
	out := make(chan bases.Subscriptor)
	go func() {
		<-time.After(d)
		out <- s.Dispose()
		close(out)
	}()
	return out
}

// UnscribeOn unsubscribes an Observable
func (s *Subscription) UnsubscribeOn(sig chan struct{}, timeout time.Duration) <-chan bases.Subscriptor {
	out := make(chan bases.Subscriptor)

	if timeout >= 0 {
		go func() {
			select {
			case <-time.After(timeout):
				return
			case <-sig:
				out <- s.Dispose()
				return
			}
		}()
	}
	go func() {
		<-sig
		out <- s.Dispose()
	}()

	return out
}<commit_msg>Refactor to comply to Subscriptor<commit_after>package subscription

import (
	"time"

	"github.com/jochasinga/grx/bases"
	"github.com/jochasinga/grx/subject"
)

// Subscription is usually returned from any subscription
type Subscription struct {
	observable    *subject.Subject
	subscribeAt   time.Time
	unsubscribeAt time.Time
}

// DefaultSubscription is a default Subscription
var DefaultSubscription = &Subscription{}

func New(fs ...func(*Subscription)) *Subscription {
	s := DefaultSubscription
	if len(fs) > 0 {
		for _, f := range fs {
			f(s)
		}
	}
	return s
}

func (s *Subscription) Stream() bases.Stream {
	return s.observable.Stream
}

// Dispose cleans up an observable and notify its unsubscribe channel and return a Subscriptor
func (s *Subscription) Dispose() {
	go func() {
		s.observable.Unsubscribe()
	}()
}

func (s *Subscription) Subscribe() bases.Subscriptor {
	s.subscribeAt = time.Now()
	return s
}

func (s *Subscription) SubscribeAt() time.Time {
	return s.subscribeAt
}

// Unsubscribe is an alias for Dispose
func (s *Subscription) Unsubscribe() bases.Subscriptor {
	s.unsubscribeAt = time.Now()
	return s
}

func (s *Subscription) UnsubscribeAt() time.Time {
	return s.unsubscribeAt
}

// UnscribeIn notify the unsubscribe channel in d duration, then return the Subscriptor
func (s *Subscription) UnsubscribeIn(d time.Duration) <-chan bases.Subscriptor {
	out := make(chan bases.Subscriptor)
	go func() {
		<-time.After(d)
		out <- s.Unsubscribe()
		close(out)
	}()
	return out
}

// UnscribeOn unsubscribes an Observable
func (s *Subscription) UnsubscribeOn(sig chan struct{}, timeout time.Duration) <-chan bases.Subscriptor {
	out := make(chan bases.Subscriptor)

	if timeout >= 0 {
		go func() {
			select {
			case <-time.After(timeout):
				return
			case <-sig:
				out <- s.Unsubscribe()
				return
			}
		}()
	}
	return out
}
{'fraction_non_alphanumeric': 0.11885361056476539, 'fraction_numerical': 0.00112391121101433, 'mean_word_length': 3.810810810810811, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3685548', 'n_tokens_mistral': 1256, 'n_tokens_neox': 1195, 'n_words': 341}