func TestProducerMessageLoop(t *testing.T) {
	expect := shared.NewExpect(t)
	mockP := getMockProducer()
	mockP.setState(PluginStateActive)
	mockP.messages = make(chan Message, 10)
	msgData := "test Message loop"
	msg := Message{
		Data: []byte(msgData),
	}

	for i := 0; i < 9; i++ {
		mockP.messages <- msg
	}
	counter := 0
	onMessage := func(msg Message) {
		expect.Equal(msgData, msg.String())
		counter++
		if counter == 9 {
			mockP.setState(PluginStateDead)
		}
	}

	mockP.messageLoop(onMessage)
	expect.Equal(9, counter)
}

func TestPluginRunState(t *testing.T) {
	expect := shared.NewExpect(t)
	pluginState := NewPluginRunState()

	expect.Equal(PluginStateDead, pluginState.GetState())

	pluginState.SetState(PluginStateWaiting)
	expect.Equal(PluginStateWaiting, pluginState.GetState())
	pluginState.SetState(PluginStateActive)
	expect.Equal(PluginStateActive, pluginState.GetState())
	pluginState.SetState(PluginStateStopping)
	expect.Equal(PluginStateStopping, pluginState.GetState())

	var wg sync.WaitGroup
	pluginState.SetWorkerWaitGroup(&wg)

	pluginState.AddWorker()
	pluginState.AddWorker()
	done := false

	go func() {
		pluginState.WorkerDone()
		pluginState.WorkerDone()
		wg.Wait()
		done = true
	}()
	// timeout for go routine.
	time.Sleep(500 * time.Millisecond)
	expect.True(done)

}

// Function reads initializes pluginConfig with predefined values and
// non-predefined values in the Settings
// Plan:
//  Create a new PluginConfig
//  Create a new shared.MarshalMap with mock key values
//  Check if they are actually added, if they are, assert the key-value correctness
func TestPluginConfigRead(t *testing.T) {
	expect := shared.NewExpect(t)
	mockPluginCfg := getMockPluginConfig()

	// create a mock MarshalMap
	testMarshalMap := shared.NewMarshalMap()
	testMarshalMap["Instances"] = 0

	mockPluginCfg.Read(testMarshalMap)
	// with 0 instance, plugin should be disabled
	expect.False(mockPluginCfg.Enable)
	// if there is no stream, then array with wildcard should be returned
	expect.Equal(mockPluginCfg.Stream, []string{"*"})

	//reset mockPluginCfg
	mockPluginCfg = getMockPluginConfig()
	testMarshalMap["ID"] = "mockId"
	testMarshalMap["Enable"] = true
	testMarshalMap["Instances"] = 2
	testMarshalMap["Stream"] = "mockStats"
	testMarshalMap["Host"] = "someHost"
	testMarshalMap["Database"] = "someDatabase"

	mockPluginCfg.Read(testMarshalMap)

	// Check for the bundled config options
	expect.Equal(mockPluginCfg.ID, "mockId")
	expect.True(mockPluginCfg.Enable)
	expect.Equal(mockPluginCfg.Instances, 2)
	expect.Equal(mockPluginCfg.Stream, []string{"mockStats"})

	// check for the miscelleneous settings key added
	expect.Equal(mockPluginCfg.GetString("Host", ""), "someHost")
	expect.Equal(mockPluginCfg.GetString("Database", ""), "someDatabase")
}

// For completeness' sake. This is exactly the same as Producer's ticket loop
func TestConsumerTickerLoop(t *testing.T) {
	expect := shared.NewExpect(t)
	mockC := getMockConsumer()
	mockC.setState(PluginStateActive)
	// accept timeroff by abs( 8 ms)
	delta := float64(8 * time.Millisecond)
	var counter = 0
	tickerLoopTimeout := 20 * time.Millisecond
	var timeRecorded time.Time
	onTimeOut := func() {
		if counter > 3 {
			mockC.setState(PluginStateDead)
			return
		}
		//this was fired as soon as the ticker started. So ignore but save the time
		if counter == 0 {
			timeRecorded = time.Now()
			counter++
			return
		}
		diff := time.Now().Sub(timeRecorded)
		deltaDiff := math.Abs(float64(tickerLoopTimeout - diff))
		expect.True(deltaDiff < delta)
		timeRecorded = time.Now()
		counter++
		return
	}

	mockC.tickerLoop(tickerLoopTimeout, onTimeOut)
	time.Sleep(2 * time.Second)
	// in anycase, the callback has to be called atleast once
	expect.True(counter > 1)
}

// For completeness' sake. This is exactly the same as Producer's control loop
func TestConsumerControlLoop(t *testing.T) {
	expect := shared.NewExpect(t)
	mockC := getMockConsumer()

	var stop bool
	var roll bool
	mockC.SetStopCallback(func() {
		stop = true
	})

	mockC.SetRollCallback(func() {
		roll = true
	})

	go mockC.ControlLoop()
	// let the go routine start
	time.Sleep(50 * time.Millisecond)
	mockC.control <- PluginControlStopConsumer
	time.Sleep(50 * time.Millisecond)
	expect.True(stop)

	go mockC.ControlLoop()
	time.Sleep(50 * time.Millisecond)
	mockC.control <- PluginControlRoll
	time.Sleep(50 * time.Millisecond)
	expect.True(roll)
}

func TestProducerControlLoop(t *testing.T) {
	expect := shared.NewExpect(t)
	mockP := getMockProducer()

	var stop bool
	var roll bool
	mockP.onStop = func() {
		stop = true
	}

	mockP.onRoll = func() {
		roll = true
	}

	go expect.NonBlocking(2*time.Second, mockP.ControlLoop)
	time.Sleep(50 * time.Millisecond)
	mockP.control <- PluginControlStopProducer // trigger stopLoop (stop expect.NonBlocking)
	time.Sleep(50 * time.Millisecond)
	expect.True(stop)

	go expect.NonBlocking(2*time.Second, mockP.ControlLoop)
	time.Sleep(50 * time.Millisecond)
	mockP.control <- PluginControlRoll // trigger rollLoop (stop expect.NonBlocking)
	time.Sleep(50 * time.Millisecond)
	expect.True(roll)

}

func TestConsumerEnqueueCopy(t *testing.T) {
	expect := shared.NewExpect(t)
	mockC := getMockConsumer()

	dataToSend := "Consumer Enqueue Data"
	distribute := func(msg Message) {
		expect.Equal(dataToSend, msg.String())
	}

	mockStream := getMockStream()
	mockP := getMockProducer()
	mockStream.AddProducer(&mockP)
	mockStream.distribute = distribute
	mockStreamID := GetStreamID("mockStream")
	StreamRegistry.Register(&mockStream, mockStreamID)

	mockC.streams = []MappedStream{
		MappedStream{
			StreamID: mockStreamID,
			Stream:   &mockStream,
		},
	}

	mockC.EnqueueCopy([]byte(dataToSend), 2)

}

func TestJSONFormatter1(t *testing.T) {
	expect := shared.NewExpect(t)

	testString := `{"a":123,"b":"string","c":[1,2,3],"d":[{"a":1}],"e":[[1,2]],"f":[{"a":1},{"b":2}],"g":[[1,2],[3,4]]}`
	msg := core.NewMessage(nil, []byte(testString), 0)
	test := newTestJSONFormatter([]interface{}{
		`findKey    :":  key        ::`,
		`findKey    :}:             : pop  : end`,
		`key        :":  findVal    :      : key`,
		`findVal    :\:: value      ::`,
		`value      :":  string     ::`,
		`value      :[:  array      : push : arr`,
		`value      :{:  findKey    : push : obj`,
		`value      :,:  findKey    :      : val`,
		`value      :}:             : pop  : val+end`,
		`string     :":  findKey    :      : esc`,
		`array      :[:  array      : push : arr`,
		`array      :{:  findKey    : push : obj`,
		`array      :]:             : pop  : end`,
		`array      :,:  arrIntVal  :      : val`,
		`array      :":  arrStrVal  ::`,
		`arrIntVal  :,:  arrIntVal  :      : val`,
		`arrIntVal  :]:             : pop  : val+end`,
		`arrStrVal  :":  arrNextStr :      : esc`,
		`arrNextStr :":  arrStrVal  ::`,
		`arrNextStr :]:             : pop  : end`,
	}, "findKey")

	result, _ := test.Format(msg)
	expect.Equal(testString, string(result))
}

// Function gets an bool value for a key or default if non-existant
// Plan: similar to TestPluginConfigGetInt
func TestPluginConfigGetBool(t *testing.T) {
	expect := shared.NewExpect(t)
	mockPluginCfg := getMockPluginConfig()

	expect.Equal(mockPluginCfg.GetBool("boolKey", false), false)
	mockPluginCfg.Settings["boolKey"] = true
	expect.Equal(mockPluginCfg.GetBool("boolKey", false), true)
}

func TestWriterAssemblySetWriter(t *testing.T) {
	expect := shared.NewExpect(t)
	mockIo := mockIoWrite{expect}
	wa := NewWriterAssembly(mockIo, mockIo.mockFlush, &mockFormatter{})

	wa.SetWriter(secondMockIoWrite{})
	length, _ := wa.writer.Write([]byte("abcde"))
	expect.Equal(length, 5)
}

func TestConsumerFuse(t *testing.T) {
	expect := shared.NewExpect(t)
	mockC := getMockConsumer()

	conf := getMockPluginConfig()
	conf.Override("Fuse", "test")

	mockC.Configure(conf)
	expect.NotNil(mockC.fuse)

	burnedCallback := false
	activeCallback := false

	mockC.SetFuseBurnedCallback(func() { burnedCallback = true })
	mockC.SetFuseActiveCallback(func() { activeCallback = true })

	go mockC.ControlLoop()

	expect.False(mockC.fuse.IsBurned())
	expect.False(burnedCallback)
	expect.False(activeCallback)

	// Check manual fuse trigger

	burnedCallback = false
	activeCallback = false

	mockC.Control() <- PluginControlFuseBurn
	time.Sleep(10 * time.Millisecond)
	expect.True(burnedCallback)
	expect.False(activeCallback)

	burnedCallback = false
	mockC.Control() <- PluginControlFuseActive
	time.Sleep(10 * time.Millisecond)
	expect.False(burnedCallback)
	expect.True(activeCallback)

	// Check automatic burn callback

	burnedCallback = false
	activeCallback = false

	mockC.fuse.Burn()
	time.Sleep(shared.SpinTimeSuspend + 100*time.Millisecond)

	expect.True(burnedCallback)
	expect.False(activeCallback)

	// Check automatic activate callback

	burnedCallback = false
	mockC.fuse.Activate()
	time.Sleep(10 * time.Millisecond)
	expect.False(burnedCallback)
	expect.True(activeCallback)
}

// Function checks if predefined value exists or not
// Plan:
//  Create a new PluginConfig
//  Add non-predefined Values
//  Check if added ones return true, and others false
func TestPluginConfigHasValue(t *testing.T) {
	expect := shared.NewExpect(t)
	mockPluginCfg := getMockPluginConfig()

	expect.False(mockPluginCfg.HasValue("aKey"))

	mockPluginCfg.Settings["aKey"] = 1
	expect.True(mockPluginCfg.HasValue("aKey"))
}

// Function gets an int value for a key or default value if non-existant
// Plan:
//  Create a new PluginConfig
//  add a key and an int value in the Settings
//  get the value back and Assert
func TestPluginConfigGetInt(t *testing.T) {
	expect := shared.NewExpect(t)
	mockPluginCfg := getMockPluginConfig()

	expect.Equal(mockPluginCfg.GetInt("intKey", 0), 0)
	mockPluginCfg.Settings["intKey"] = 2
	expect.Equal(mockPluginCfg.GetInt("intKey", 0), 2)

}

func TestWriterAssemblySetValidator(t *testing.T) {
	expect := shared.NewExpect(t)
	mockIo := mockIoWrite{expect}
	wa := NewWriterAssembly(mockIo, mockIo.mockFlush, &mockFormatter{})
	validator := func() bool {
		return true
	}
	wa.SetValidator(validator)
	expect.True(wa.validate())
}

func TestStreamRegistryRegister(t *testing.T) {
	expect := shared.NewExpect(t)
	mockSRegistry := getMockStreamRegistry()

	streamName := "testStream"
	mockStream := getMockStream()
	mockSRegistry.Register(&mockStream, GetStreamID(streamName))

	expect.NotNil(mockSRegistry.GetStream(GetStreamID(streamName)))
}

// Function gets the string value for a key or if the key/value(?) doesn't
// exist, returns the default value
// Plan:
//  Create a new PluginConfig
//  For a random key, check if the default value is returned
//  Add a key with a value
//  Asserts the string returned for the key is correct
func TestPluginConfigGetString(t *testing.T) {
	expect := shared.NewExpect(t)
	mockPluginCfg := getMockPluginConfig()

	//check for non-existant key
	expect.Equal(mockPluginCfg.GetString("aKey", "default"), "default")

	mockPluginCfg.Settings["aKey"] = "aValue"
	expect.Equal(mockPluginCfg.GetString("aKey", "default"), "aValue")
}

func TestStreamRegistryIsStreamRegistered(t *testing.T) {
	expect := shared.NewExpect(t)
	mockSRegistry := getMockStreamRegistry()

	mockStreamID := GetStreamID("testStream")

	expect.False(mockSRegistry.IsStreamRegistered(mockStreamID))
	// TODO: Get a real stream and test with that
	mockSRegistry.streams[mockStreamID] = &StreamBase{}
	expect.True(mockSRegistry.IsStreamRegistered(mockStreamID))
}

func TestStreamRegistryForEachStream(t *testing.T) {
	expect := shared.NewExpect(t)
	mockSRegistry := getMockStreamRegistry()

	callback := func(streamID MessageStreamID, stream Stream) {
		expect.Equal(streamID, GetStreamID("testRegistry"))
	}

	mockSRegistry.streams[GetStreamID("testRegistry")] = &StreamBase{}
	mockSRegistry.ForEachStream(callback)
}

func TestStreamRegistryGetStreamByName(t *testing.T) {
	expect := shared.NewExpect(t)
	mockSRegistry := getMockStreamRegistry()

	streamName := mockSRegistry.GetStreamByName("testStream")
	expect.Equal(streamName, nil)

	mockStreamID := GetStreamID("testStream")
	// TODO: Get a real stream and test with that
	mockSRegistry.streams[mockStreamID] = &StreamBase{}
	expect.Equal(mockSRegistry.GetStreamByName("testStream"), &StreamBase{})
}

func TestConsumerControl(t *testing.T) {
	expect := shared.NewExpect(t)
	mockC := getMockConsumer()
	mockC.control = make(chan PluginControl, 2)

	ctrlChan := mockC.Control()
	ctrlChan <- PluginControlRoll
	ctrlChan <- PluginControlStopConsumer

	expect.Equal(PluginControlRoll, <-mockC.control)
	expect.Equal(PluginControlStopConsumer, <-mockC.control)
}