func (c *EventCollector) CheckRetrievedEventsForNewsroom(pastEvents []*model.Event) (
	[]*model.Event, error) {
	existingFiltererNewsroomAddr := c.getExistingNewsroomFilterers()
	existingWatcherNewsroomAddr := c.getExistingNewsroomWatchers()
	existingFiltererMultiSigAddr := c.getExistingMultiSigFilterers()
	existingWatcherMultiSigAddr := c.getExistingMultiSigWatchers()
	watchersToAdd := map[common.Address]model.ContractWatchers{}
	filterersToAdd := map[common.Address]model.ContractFilterers{}
	eventsToAdd := []*model.Event{}
	for _, event := range pastEvents {
		var multiSigAddr common.Address
		if event.EventType() == "ContractInstantiation" && event.ContractName() == "MultiSigWalletFactoryContract" {
			addr, ok := event.EventPayload()["Instantiation"].(common.Address)
			if !ok {
				return eventsToAdd, fmt.Errorf("Cannot get multiSigAddr from eventpayload")
			}
			multiSigAddr = addr
		} else if event.EventType() == "Application" {
			addr, ok := event.EventPayload()["Applicant"].(common.Address)
			if !ok {
				return eventsToAdd, fmt.Errorf("Cannot get applicant from eventpayload")
			}
			multiSigAddr = addr
		}
		if multiSigAddr.String() != "" {
			if _, ok := existingFiltererMultiSigAddr[multiSigAddr]; !ok {
				newFilterer := filterer.NewMultiSigWalletContractFilterers(multiSigAddr)
				filterersToAdd[multiSigAddr] = newFilterer
				existingFiltererMultiSigAddr[multiSigAddr] = true
			}
			if _, ok := existingWatcherMultiSigAddr[multiSigAddr]; !ok {
				newWatcher := watcher.NewMultiSigWalletContractWatchers(multiSigAddr)
				watchersToAdd[multiSigAddr] = newWatcher
				existingWatcherMultiSigAddr[multiSigAddr] = true
			}
		}
		var newsroomAddr common.Address
		if event.EventType() == "ContractInstantiation" && event.ContractName() == "NewsroomFactory" {
			addr, ok := event.EventPayload()["Instantiation"].(common.Address)
			if !ok {
				return eventsToAdd, fmt.Errorf("Cannot get newsroomAddr from eventpayload")
			}
			newsroomAddr = addr
		} else if event.EventType() == "Application" {
			addr, ok := event.EventPayload()["ListingAddress"].(common.Address)
			if !ok {
				return eventsToAdd, fmt.Errorf("Cannot get newsroomAddr from eventpayload")
			}
			newsroomAddr = addr
		}
		if newsroomAddr.String() != "" {
			if _, ok := existingFiltererNewsroomAddr[newsroomAddr]; !ok {
				newFilterer := filterer.NewNewsroomContractFilterers(newsroomAddr)
				filterersToAdd[newsroomAddr] = newFilterer
				existingFiltererNewsroomAddr[newsroomAddr] = true
			}
			if _, ok := existingWatcherNewsroomAddr[newsroomAddr]; !ok {
				newWatcher := watcher.NewNewsroomContractWatchers(newsroomAddr)
				watchersToAdd[newsroomAddr] = newWatcher
				existingWatcherNewsroomAddr[newsroomAddr] = true
			}
		}
	}
	var e error
	for addr, watcher := range watchersToAdd {
		if watcher != nil {
			log.Infof("Adding new watcher for %v", addr.Hex())
			e = c.AddWatchers(watcher)
			if e != nil {
				log.Errorf("Error adding new filterer: err: %v", e)
			}
		} else {
			log.Infof("Not adding %v to watchers because it was removed.", addr.Hex())
		}
	}
	if len(filterersToAdd) > 0 {
		newFilts := make([]model.ContractFilterers, len(filterersToAdd))
		ind := 0
		for addr, filt := range filterersToAdd {
			log.Infof("Adding new filterer for %v", addr.Hex())
			newFilts[ind] = filt
			ind++
			e = c.AddFilterers(filt)
			if e != nil {
				log.Errorf("Error adding new filterer: err: %v", e)
			}
		}
		r, err := c.retrieveEvents(newFilts, false)
		if err != nil {
			return eventsToAdd, errors.WithMessage(err, "error retrieving new Newsroom or MultiSig events")
		}
		eventsToAdd = append(eventsToAdd, r.PastEvents...)
	}
	return eventsToAdd, nil
}
CheckRetrievedEventsForNewsroom checks pastEvents for TCR events that may
include new newsroom events, creates new Newsroom filterers and watchers
upon valid events, filters for events, and then returns those events.
package eventcollector  
import (
	"fmt"
	"sync"
	"time"
	log "github.com/golang/glog"
	"github.com/pkg/errors"
	"github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/accounts/abi/bind"
	"github.com/ethereum/go-ethereum/common"
	"github.com/joincivil/civil-events-crawler/pkg/generated/filterer"
	"github.com/joincivil/civil-events-crawler/pkg/generated/watcher"
	"github.com/joincivil/civil-events-crawler/pkg/listener"
	"github.com/joincivil/civil-events-crawler/pkg/model"
	"github.com/joincivil/civil-events-crawler/pkg/pubsub"
	cerrors "github.com/joincivil/go-common/pkg/errors"
	"github.com/joincivil/go-common/pkg/eth"
)
const (
	blockHeaderExpirySecs = 60 * 5  
	wsPingDelaySecs = 60 * 2  
	defaultPollingIntervalSecs = 60  
)
type Config struct {
	Chain               ethereum.ChainReader
	HTTPClient          bind.ContractBackend
	WsClient            bind.ContractBackend
	WsEthURL            string
	ErrRep              cerrors.ErrorReporter
	Filterers           []model.ContractFilterers
	Watchers            []model.ContractWatchers
	RetrieverPersister  model.RetrieverMetaDataPersister
	ListenerPersister   model.ListenerMetaDataPersister
	EventDataPersister  model.EventDataPersister
	Triggers            []Trigger
	StartBlock          uint64
	CrawlerPubSub       *pubsub.CrawlerPubSub
	PollingEnabled      bool
	PollingIntervalSecs int
	PreemptSecs         *int
}
func NewEventCollector(config *Config) *EventCollector {
	eventcollector := &EventCollector{
		chain:               config.Chain,
		retryChain:          eth.RetryChainReader{ChainReader: config.Chain},
		httpClient:          config.HTTPClient,
		wsClient:            config.WsClient,
		wsEthURL:            config.WsEthURL,
		errRep:              config.ErrRep,
		filterers:           config.Filterers,
		watchers:            config.Watchers,
		retrieverPersister:  config.RetrieverPersister,
		listenerPersister:   config.ListenerPersister,
		eventDataPersister:  config.EventDataPersister,
		triggers:            config.Triggers,
		startBlock:          config.StartBlock,
		startChan:           make(chan struct{}),
		shutdownChan:        make(chan struct{}),
		listenerStopChan:    make(chan struct{}),
		listenerErrChan:     make(chan error),
		retrieverStopChan:   make(chan struct{}),
		retrieverErrChan:    make(chan error),
		crawlerPubSub:       config.CrawlerPubSub,
		pollingEnabled:      config.PollingEnabled,
		pollingIntervalSecs: config.PollingIntervalSecs,
		preemptSecs:         config.PreemptSecs,
	}
	return eventcollector
}
type EventCollector struct {
	chain ethereum.ChainReader
	retryChain eth.RetryChainReader
	httpClient bind.ContractBackend
	wsClient bind.ContractBackend
	wsEthURL string
	errRep cerrors.ErrorReporter
	triggers []Trigger
	filterers []model.ContractFilterers
	watchers []model.ContractWatchers
	retrieverPersister model.RetrieverMetaDataPersister
	listenerPersister model.ListenerMetaDataPersister
	eventDataPersister model.EventDataPersister
	listen *listener.EventListener
	startBlock uint64
	startChan chan struct{}
	shutdownChan chan struct{}
	retrieverStopChan chan struct{}
	retrieverErrChan chan error
	listenerStopChan chan struct{}
	listenerErrChan chan error
	mutex sync.Mutex
	headerCache *eth.BlockHeaderCache
	crawlerPubSub *pubsub.CrawlerPubSub
	pollingEnabled bool
	pollingIntervalSecs int
	preemptSecs *int
}
func (c *EventCollector) StartChan() chan struct{} {
	defer c.mutex.Unlock()
	c.mutex.Lock()
	return c.startChan
}
func (c *EventCollector) StartCollection() error {
	defer c.resetCollector()
	var once sync.Once
	var err error
	var l *listener.EventListener
	pd := c.preemptDuration()
	for {
		c.retrieverStopChan = make(chan struct{})
		err = c.runRetrieverLoop(c.pollingEnabled)
		if err != nil {
			once.Do(func() { c.sendStartSignal() })
			return errors.Wrap(err, "startcollection.runRetrieverLoop")
		}
		if !c.pollingEnabled && c.isListenerEnabled() {
			c.listenerStopChan = make(chan struct{})
			l, err = c.startListenerLoop()
			if err != nil {
				once.Do(func() { c.sendStartSignal() })
				log.Errorf("Error starting listener: err: %v", err)
				return errors.Wrap(err, "error starting listener in startcol")
			}
			c.mutex.Lock()
			c.listen = l
			c.mutex.Unlock()
			log.Infof("Listeners enabled")
			once.Do(func() { c.sendStartSignal() })
		}
		if c.pollingEnabled {
			once.Do(func() { c.sendStartSignal() })
			log.Infof("Polling via retrievers only")
		}
		select {
		case <-time.After(pd):
			log.Infof("Premptive restart...")
			c.resetCollector()
		case err = <-c.retrieverErrChan:
			err = errors.Cause(err)
			log.Errorf("Received error from retriever, resetting collector: err: %v", err)
			c.errRep.Error(err, nil)
			c.resetCollector()
		case err = <-c.listenerErrChan:
			err = errors.Cause(err)
			log.Errorf("Received error from listener, resetting collector: err: %v", err)
			c.errRep.Error(err, nil)
			c.resetCollector()
		case <-c.shutdownChan:
			log.Infof("Received shutdownChan signal, stopping collection")
			return nil
		}
		log.Infof("Restarting collection loop")
	}
}
func (c *EventCollector) StopCollection(unsubWatchers bool) error {
	defer c.mutex.Unlock()
	c.mutex.Lock()
	var err error
	if c.crawlerPubSub != nil {
		err = c.crawlerPubSub.StopPublishers()
		if err != nil {
			return err
		}
	}
	if c.listen != nil {
		err = c.listen.Stop(unsubWatchers)
	}
	if c.shutdownChan != nil {
		close(c.shutdownChan)
	}
	return err
}