wip

Author: alecholmes

cshared_test.go

@@ -80,12 +80,7 @@ func TestInputCallbackCtrlC(t *testing.T) {
 
 	ptr := unsafe.Pointer(nil)
 
-	// prepare channel for input explicitly.
-	//prepareInputCollector(false)
-	//require.NoError(t, inst.inputPreRun())
-
 	go func() {
-		//FLBPluginInputCallback(&ptr, nil)
 		inst.callback(&ptr, nil)
 		close(cdone)
 	}()
@@ -135,7 +130,6 @@ func TestInputCallbackDangle(t *testing.T) {
 		ticker := time.NewTicker(collectInterval)
 		defer ticker.Stop()
 
-		//FLBPluginInputCallback(&ptr, nil)
 		inst.callback(&ptr, nil)
 		for {
 			select {
@@ -205,7 +199,6 @@ func TestInputCallbackInfinite(t *testing.T) {
 		for {
 			select {
 			case <-ticker.C:
-				//FLBPluginInputCallback(&ptr, nil)
 				inst.callback(&ptr, nil)
 				if ptr != nil {
 					close(cdone)

plugin.go

@@ -214,6 +214,9 @@ func (p *pluginInstance) withConfigLoaderProvider(provider configLoaderProvider)
 }
 
 // init initializes a newly created plugin instance.
+// This returns an error if the plugin if not in a new state or cannot initialize.
+// For input and output plugins, this moves the plugin to an initialized state unless there is an init error.
+// For custom plugins, this moves the plugin to a runnable state unless there is an init error.
 func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -296,6 +299,7 @@ func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 	return nil
 }
 
+// inputPreRun transitions an initialized input plugin into runnable state.
 func (p *pluginInstance) inputPreRun() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -331,6 +335,7 @@ func (p *pluginInstance) inputPreRun() error {
 	return nil
 }
 
+// outputPreRun transitions an output plugin into runnable state.
 func (p *pluginInstance) outputPreRun() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -365,6 +370,10 @@ func (p *pluginInstance) outputPreRun() error {
 	return nil
 }
 
+// callback consumes up to maxBufferedMessages message from the plugin's message channel,
+// returning early if the plugin is shutdown.
+// Consumed messages are marshaled into msgpack bytes and the contents and length of contents
+// set in the respective data and csize input variables.
 func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
 	p.mu.Lock()
 	if p.state != instanceStateRunnable {
@@ -402,9 +411,9 @@ func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
 			// fluent-bit to kick in before any remaining data has not been GC'ed
 			// causing a sigsegv.
 			defer runtime.GC()
-			loop = 0
+			loop = 0 // Exit the for loop on plugin shutdown
 		default:
-			loop = 0
+			loop = 0 // Exit the for loop if there are no messages to consume
 		}
 	}
 
@@ -420,6 +429,9 @@ func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
 	return input.FLB_OK
 }
 
+// stop stops the plugin, freeing resources and returning it to initialized state.
+// Calling stop will signal callbacks to return via the plugin's context, then wait for
+// callbacks to finish and return before freeing resources and returning.
 func (p *pluginInstance) stop() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -446,6 +458,7 @@ func (p *pluginInstance) stop() error {
 	return nil
 }
 
+// resume restarts plugins, running pre-run functions for input and output plugins.
 func (p *pluginInstance) resume() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -454,14 +467,24 @@ func (p *pluginInstance) resume() error {
 		return fmt.Errorf("cannot resume plugin in state %q", p.state)
 	}
 
-	if err := p.inputPreRun(); err != nil {
-		return err
+	if p.meta.input != nil {
+		if err := p.inputPreRun(); err != nil {
+			return err
+		}
+	} else if p.meta.output != nil {
+		if err := p.outputPreRun(); err != nil {
+			return err
+		}
 	}
 
-	p.state = instanceStateInitialized
+	p.state = instanceStateRunnable
+
 	return nil
 }
 
+// stop prepares to stop an output plugin.
+// Calling outputPreExit will signal callbacks to return via the plugin's context, then wait for
+// callbacks to finish and return before freeing resources and returning.
 func (p *pluginInstance) outputPreExit() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -484,6 +507,7 @@ func (p *pluginInstance) outputPreExit() error {
 	p.state = instanceStatePreExit
 	p.runCtx = nil
 	p.runCancel = nil
+	p.msgChannel = nil
 
 	return nil
 }