changes to plugin and commands to accomodate changes

cmd/compose.go

@@ -62,7 +62,7 @@ var composeCmd = &cobra.Command{
         rank = "0"
     [exporter.transformer2]
         path = "path/to/transformer2"
-        type = "eth_event"
+        type = "eth_generic"
         repository = "github.com/account/repo"
         migrations = "db/migrations"
         rank = "0"
@@ -91,7 +91,8 @@ from it and loaded into and executed over by the appropriate watcher.
 The type of watcher that the transformer works with is specified using the 
 type variable for each transformer in the config. Currently there are watchers 
 of event data from an eth node (eth_event) and storage data from an eth node 
-(eth_storage).
+(eth_storage), and a more generic interface for accepting omni pkg based transformers
+which can perform both event watching and public method polling.
 
 Transformers of different types can be ran together in the same command using a 
 single config file or in separate command instances using different config files

cmd/composeAndExecute.go

@@ -62,7 +62,7 @@ var composeAndExecuteCmd = &cobra.Command{
         rank = "0"
     [exporter.transformer2]
         path = "path/to/transformer2"
-        type = "eth_event"
+        type = "eth_generic"
         repository = "github.com/account/repo"
         migrations = "db/migrations"
         rank = "2"
@@ -91,7 +91,8 @@ from it and loaded into and executed over by the appropriate watcher.
 The type of watcher that the transformer works with is specified using the 
 type variable for each transformer in the config. Currently there are watchers 
 of event data from an eth node (eth_event) and storage data from an eth node 
-(eth_storage).
+(eth_storage), and a more generic interface for accepting omni pkg based transformers
+which can perform both event watching and public method polling.
 
 Transformers of different types can be ran together in the same command using a 
 single config file or in separate command instances using different config files
@@ -150,7 +151,7 @@ func composeAndExecute() {
 	}
 
 	// Use the Exporters export method to load the EventTransformerInitializer and StorageTransformerInitializer sets
-	ethEventInitializers, ethStorageInitializers := exporter.Export()
+	ethEventInitializers, ethStorageInitializers, genericInitializers := exporter.Export()
 
 	// Setup bc and db objects
 	blockChain := getBlockChain()
@@ -173,6 +174,13 @@ func composeAndExecute() {
 		wg.Add(1)
 		go watchEthStorage(&sw, &wg)
 	}
+
+	if len(genericInitializers) > 0 {
+		gw := watcher.NewGenericWatcher(&db, blockChain)
+		gw.AddTransformers(genericInitializers)
+		wg.Add(1)
+		go genericWatching(&gw, &wg)
+	}
 	wg.Wait()
 }
 

cmd/execute.go

@@ -100,7 +100,7 @@ func execute() {
 	}
 
 	// Use the Exporters export method to load the EventTransformerInitializer and StorageTransformerInitializer sets
-	ethEventInitializers, ethStorageInitializers := exporter.Export()
+	ethEventInitializers, ethStorageInitializers, genericInitializers := exporter.Export()
 
 	// Setup bc and db objects
 	blockChain := getBlockChain()
@@ -123,6 +123,13 @@ func execute() {
 		wg.Add(1)
 		go watchEthStorage(&sw, &wg)
 	}
+
+	if len(genericInitializers) > 0 {
+		gw := watcher.NewGenericWatcher(&db, blockChain)
+		gw.AddTransformers(genericInitializers)
+		wg.Add(1)
+		go genericWatching(&gw, &wg)
+	}
 	wg.Wait()
 }
 
@@ -132,7 +139,7 @@ func init() {
 }
 
 type Exporter interface {
-	Export() ([]transformer.EventTransformerInitializer, []transformer.StorageTransformerInitializer)
+	Export() ([]transformer.EventTransformerInitializer, []transformer.StorageTransformerInitializer, []transformer.GenericTransformerInitializer)
 }
 
 func watchEthEvents(w *watcher.EventWatcher, wg *syn.WaitGroup) {
@@ -157,7 +164,7 @@ func watchEthEvents(w *watcher.EventWatcher, wg *syn.WaitGroup) {
 
 func watchEthStorage(w *watcher.StorageWatcher, wg *syn.WaitGroup) {
 	defer wg.Done()
-	// Execute over the StorageTransformerInitializer set using the watcher
+	// Execute over the StorageTransformerInitializer set using the storage watcher
 	log.Info("executing storage transformers")
 	ticker := time.NewTicker(pollingInterval)
 	defer ticker.Stop()
@@ -168,3 +175,17 @@ func watchEthStorage(w *watcher.StorageWatcher, wg *syn.WaitGroup) {
 		}
 	}
 }
+
+func genericWatching(w *watcher.GenericWatcher, wg *syn.WaitGroup) {
+	defer wg.Done()
+	// Execute over the GenericTransformerInitializer set using the generic watcher
+	log.Info("executing generic transformers")
+	ticker := time.NewTicker(pollingInterval)
+	defer ticker.Stop()
+	for range ticker.C {
+		err := w.Execute(nil)
+		if err != nil {
+			// TODO Handle watcher errors in execute
+		}
+	}
+}