Fix seg fault in CharonClient due to out of order initialization.

Specter/src/CMakeLists.txt

@@ -75,6 +75,15 @@ target_sources(Specter PRIVATE
     Specter/Physics/Caen/CompassOnlineSource.h
     Specter/Physics/Caen/CompassRun.cpp
     Specter/Physics/Caen/CompassRun.h
+    Specter/Physics/nscldaq/CharonOnlineSource.h
+    Specter/Physics/nscldaq/CharonOnlineSource.cpp
+    Specter/Physics/nscldaq/CharonClient.h
+    Specter/Physics/nscldaq/CharonClient.cpp
+    Specter/Physics/nscldaq/Unpackers/Unpacker.h
+    Specter/Physics/nscldaq/Unpackers/CaenUnpacker.h
+    Specter/Physics/nscldaq/Unpackers/CaenUnpacker.cpp
+    Specter/Physics/nscldaq/Unpackers/MesyTecUnpacker.h
+    Specter/Physics/nscldaq/Unpackers/MesyTecUnpacker.cpp
     Specter/Editor/EditorLayer.cpp
     Specter/Editor/EditorLayer.h
     Specter/Editor/FileDialog.cpp
@@ -106,6 +115,7 @@ target_sources(Specter PRIVATE
     Specter/Utils/Functions.h
     Specter/Utils/Functions.cpp
     Specter/Utils/RandomGenerator.h
+    Specter/Utils/ThreadSafeQueue.h
 )
 
 #ImPlot sources

Specter/src/Specter/Editor/SourceDialog.cpp

@@ -32,7 +32,7 @@ namespace Specter {
 		static bool onlineFlag = false;
 		static bool offlineFlag = false;
 		static std::vector<DataSource::SourceType> availTypes = { DataSource::SourceType::CompassOnline, DataSource::SourceType::CompassOffline, DataSource::SourceType::DaqromancyOnline,
-																  DataSource::SourceType::DaqromancyOffline };
+																  DataSource::SourceType::DaqromancyOffline, DataSource::SourceType::CharonOnline };
 		
 		if (m_openFlag)
 		{
@@ -78,6 +78,7 @@ namespace Specter {
 				{
 					m_bitflags = m_bitflags ^ CompassHeaders::EnergyCalibrated;
 				}
+				ImGui::InputInt("Coinc. Window (ps)", &m_chosenWindow);
 			}
 			else if (m_chosenType == DataSource::SourceType::CompassOffline)
 			{
@@ -90,11 +91,13 @@ namespace Specter {
 				auto temp = m_fileDialog.RenderFileDialog();
 				if (!temp.first.empty() && temp.second == FileDialog::Type::OpenDir)
 					m_chosenLocation = temp.first;
+				ImGui::InputInt("Coinc. Window (ps)", &m_chosenWindow);
 			}
 			else if (m_chosenType == DataSource::SourceType::DaqromancyOnline)
 			{
 				ImGui::InputText("Hostname", &m_chosenLocation);
 				ImGui::InputText("Port", &m_chosenPort);
+				ImGui::InputInt("Coinc. Window (ps)", &m_chosenWindow);
 			}
 			else if (m_chosenType == DataSource::SourceType::DaqromancyOffline)
 			{
@@ -107,22 +110,19 @@ namespace Specter {
 				auto temp = m_fileDialog.RenderFileDialog();
 				if (!temp.first.empty() && temp.second == FileDialog::Type::OpenDir)
 					m_chosenLocation = temp.first;
-			}
 				ImGui::InputInt("Coinc. Window (ps)", &m_chosenWindow);
-
+			}
+			else if (m_chosenType == DataSource::SourceType::CharonOnline)
+			{
+				ImGui::InputText("Hostname", &m_chosenLocation);
+				ImGui::InputText("Port", &m_chosenPort);
+			}
 
 			if (ImGui::Button("Ok"))
 			{
-				if (m_chosenType == DataSource::SourceType::CompassOffline || m_chosenType == DataSource::SourceType::DaqromancyOffline)
-				{
-					PhysicsStartEvent event(m_chosenLocation, m_chosenType, m_chosenWindow, m_chosenPort, false, 0U);
+				PhysicsStartEvent event(m_chosenLocation, m_chosenType, m_chosenWindow, m_chosenPort, m_bitflags);
 				Application::Get().OnEvent(event);
-				}
-				else if (m_chosenType == DataSource::SourceType::CompassOnline || m_chosenType == DataSource::SourceType::DaqromancyOnline)
-				{
-					PhysicsStartEvent event(m_chosenLocation, m_chosenType, m_chosenWindow, m_chosenPort, true, m_bitflags);
-					Application::Get().OnEvent(event);
-				}
+
 				ImGui::CloseCurrentPopup();
 			}
 			ImGui::SameLine();

Specter/src/Specter/Events/PhysicsEvent.h

@@ -21,15 +21,14 @@ namespace Specter {
 	{
 	public:
 		//Bitflags is a final option for random crap needed for a source. Currently used for compass online to indicate header state.
-		PhysicsStartEvent(const std::string& loc, DataSource::SourceType type, uint64_t window, const std::string& port = "51489", bool sortFlag=false, uint16_t bitflags = 0) :
-			m_sourceLocation(loc), m_port(port), m_sourceType(type), m_coincidenceWindow(window), m_sortFlag(sortFlag), m_bitflags(bitflags)
+		PhysicsStartEvent(const std::string& loc, DataSource::SourceType type, uint64_t window, const std::string& port = "51489", uint16_t bitflags = 0) :
+			m_sourceLocation(loc), m_port(port), m_sourceType(type), m_coincidenceWindow(window), m_bitflags(bitflags)
 		{}
 
 		inline const std::string GetSourceLocation() const { return m_sourceLocation; }
         inline const std::string GetSourcePort() const { return m_port; }
 		inline const DataSource::SourceType GetSourceType() const { return m_sourceType; }
 		inline const uint64_t GetCoincidenceWindow() const { return m_coincidenceWindow; }
-		inline const bool GetSortFlag() const { return m_sortFlag; }
 		inline const uint16_t GetBitFlags() const { return m_bitflags; }
 
 		std::string ToString() const override
@@ -45,7 +44,6 @@ namespace Specter {
         std::string m_port;
 		DataSource::SourceType m_sourceType;
 		uint64_t m_coincidenceWindow;
-		bool m_sortFlag;
 		uint16_t m_bitflags;
 	};
 

Specter/src/Specter/ImGui/ImGuiLayer.cpp

@@ -126,7 +126,7 @@ namespace Specter {
 	void ImGuiLayer::OnImGuiRender()
 	{
 		//Demo's used to figure out how to do things.
-		//Should not be on for actual NavProject for 
+		//Should not be on for actual SpecProject for 
 		//real use
 		//static bool show = true;
 		//ImGui::ShowDemoWindow(&show);

Specter/src/Specter/Physics/Caen/CompassOnlineSource.cpp

@@ -22,9 +22,10 @@
 
 namespace Specter {
 
-	CompassOnlineSource::CompassOnlineSource(const std::string& hostname, const std::string& port, uint16_t header) :
-		DataSource(), m_bufferIter(nullptr), m_bufferEnd(nullptr), m_header(header)
+	CompassOnlineSource::CompassOnlineSource(const std::string& hostname, const std::string& port, uint16_t header, uint64_t coincidenceWindow) :
+		DataSource(coincidenceWindow), m_bufferIter(nullptr), m_bufferEnd(nullptr), m_header(header)
 	{
+		m_eventBuilder.SetSortFlag(true);
 		InitConnection(hostname, port);
 	}
 
@@ -52,15 +53,14 @@ namespace Specter {
 		}
 	}
 
-	const SpecData& CompassOnlineSource::GetData()
+	void CompassOnlineSource::ProcessData()
 	{
 		SPEC_PROFILE_FUNCTION();
 		size_t range = m_bufferEnd - m_bufferIter; //how much buffer we have left
 		if (!IsValid())
 		{
 			SPEC_ERROR("Attempting to access invalid source at CompassOnlineSource!");
-			m_datum = SpecData();
-			return m_datum;
+			return;
 		}
 		else if (m_bufferIter == nullptr || range < m_datasize || m_bufferIter == m_bufferEnd) //If no buffer/buffer completely used/buffer fragmented fill 
 		{
@@ -71,8 +71,7 @@ namespace Specter {
 			GetHit();
 		else
 		{
-			m_datum = SpecData();
-			return m_datum;
+			return;
 		}
 
 		m_datum.longEnergy = m_currentHit.energy;
@@ -81,7 +80,10 @@ namespace Specter {
 		m_datum.timestamp = m_currentHit.timestamp;
 		m_datum.id = Utilities::GetBoardChannelUUID(m_currentHit.board, m_currentHit.channel);
 
-		return m_datum;
+		if(m_eventBuilder.AddDatum(m_datum))
+		{
+			m_isEventReady = true;
+		}
 	}
 
 	void CompassOnlineSource::FillBuffer()

Specter/src/Specter/Physics/Caen/CompassOnlineSource.h

@@ -36,10 +36,15 @@ namespace Specter {
 	class CompassOnlineSource : public DataSource
 	{
 	public:
-		CompassOnlineSource(const std::string& hostname, const std::string& port, uint16_t header);
+		CompassOnlineSource(const std::string& hostname, const std::string& port, uint16_t header, uint64_t coincidenceWindow);
 		virtual ~CompassOnlineSource() override;
 
-		virtual const SpecData& GetData() override;
+		virtual void ProcessData() override;
+		virtual const std::vector<SpecEvent>& GetEvents() override
+		{
+			m_isEventReady = false;
+			return m_eventBuilder.GetReadyEvents();
+		}
 
 	private:
 		void InitConnection(const std::string& hostname, const std::string& port);

Specter/src/Specter/Physics/Caen/CompassRun.cpp

@@ -26,13 +26,8 @@
 
 namespace Specter {
 	
-	CompassRun::CompassRun() :
-		DataSource(), m_directory(""), m_startIndex(0)
-	{
-	}
-	
-	CompassRun::CompassRun(const std::string& dir) :
-		DataSource(), m_directory(dir), m_startIndex(0)
+	CompassRun::CompassRun(const std::string& dir, uint64_t coincidenceWindow) :
+		DataSource(coincidenceWindow), m_directory(dir), m_startIndex(0)
 	{
 		CollectFiles();
 	}
@@ -129,32 +124,32 @@ namespace Specter {
 		return true;
 	}
 
-	const SpecData& CompassRun::GetData()
+	void CompassRun::ProcessData()
 	{
 		SPEC_PROFILE_FUNCTION();
 		if(!IsValid())
 		{
 			SPEC_ERROR("Trying to access CompassRun data when invalid, bug detected!");
-			m_datum = SpecData();
-			return m_datum;
+			return;
 		}
 
 		if (!GetHitsFromFiles())
 		{
 			m_validFlag = false;
-			m_datum = SpecData();
+			return;
 		}
-		else
-		{
+		
 		//Convert data from CoMPASS format to universal Specter format.
 		m_datum.longEnergy = m_hit.energy;
 		m_datum.shortEnergy = m_hit.energyShort;
 		m_datum.calEnergy = m_hit.energyCalibrated;
 		m_datum.timestamp = m_hit.timestamp;
 		m_datum.id = Utilities::GetBoardChannelUUID(m_hit.board, m_hit.channel);
-		}
 
-		return m_datum;
+		if(m_eventBuilder.AddDatum(m_datum))
+		{
+			m_isEventReady = true;
+		}
 	}
 
 }

Specter/src/Specter/Physics/Caen/CompassRun.h

@@ -33,12 +33,15 @@ namespace Specter {
 	
 	class CompassRun : public DataSource
 	{
-	
 	public:
-		CompassRun();
-		CompassRun(const std::string& dir);
+		CompassRun(const std::string& dir, uint64_t coincidenceWindow);
 		virtual ~CompassRun();
-		virtual const SpecData& GetData() override;
+		virtual void ProcessData() override;
+		virtual const std::vector<SpecEvent>& GetEvents() override
+		{
+			m_isEventReady = false;
+			return m_eventBuilder.GetReadyEvents();
+		}
 		inline void SetDirectory(const std::string& dir) { m_directory = dir; CollectFiles(); }
 		inline void SetShiftMap(const std::string& filename) { m_smap.SetFile(filename); }
 	
@@ -58,6 +61,7 @@ namespace Specter {
 		CompassHit m_hit;
 	
 		unsigned int m_totalHits;
+
 	};
 
 }

Specter/src/Specter/Physics/Daqromancy/DYFileSource.cpp

@@ -2,8 +2,8 @@
 
 namespace Specter {
 
-	DYFileSource::DYFileSource(const std::string& directory) :
-		DataSource(), m_directory(directory)
+	DYFileSource::DYFileSource(const std::string& directory, uint64_t coicidenceWindow) :
+		DataSource(coicidenceWindow), m_directory(directory)
 	{
 		CollectFiles();
 	}
@@ -88,28 +88,26 @@ namespace Specter {
 		return true;
 	}
 
-	const SpecData& DYFileSource::GetData()
+	void DYFileSource::ProcessData()
 	{
 		if (!IsValid())
 		{
 			SPEC_ERROR("Trying to access DYFileSource data when invalid, bug detected!");
-			m_datum = SpecData();
-			return m_datum;
+			return;
 		}
 
 		if (!GetNextHit())
 		{
 			m_validFlag = false;
-			m_datum = SpecData();
 		}
-		else
-		{
 		//Convert data from Daqromancy format to universal Specter format.
 		m_datum.longEnergy = m_dyHit.energy;
 		m_datum.shortEnergy = m_dyHit.energyShort;
 		m_datum.timestamp = m_dyHit.timestamp;
 		m_datum.id = Utilities::GetBoardChannelUUID(m_dyHit.board, m_dyHit.channel);
-		}
-		return m_datum;
+		if(m_eventBuilder.AddDatum(m_datum))
+		{
+			m_isEventReady = true;
+		}
 	}
 }

Specter/src/Specter/Physics/Daqromancy/DYFileSource.h

@@ -10,10 +10,15 @@ namespace Specter {
 	class DYFileSource : public DataSource
 	{
 	public:
-		DYFileSource(const std::string& directory);
+		DYFileSource(const std::string& directory, uint64_t coincidenceWindow);
 		virtual ~DYFileSource();
 
-		virtual const SpecData& GetData() override;
+		virtual void ProcessData() override;
+		virtual const std::vector<SpecEvent>& GetEvents() override
+		{
+			m_isEventReady = false;
+			return m_eventBuilder.GetReadyEvents();
+		}
 
 	private:
 		void CollectFiles();

Specter/src/Specter/Physics/Daqromancy/DYOnlineSource.cpp

@@ -2,9 +2,10 @@
 
 namespace Specter {
 
-	DYOnlineSource::DYOnlineSource(const std::string& hostname, const std::string& port) :
-		DataSource(), m_clientConnection(hostname, port)
+	DYOnlineSource::DYOnlineSource(const std::string& hostname, const std::string& port, uint64_t coincidenceWindow) :
+		DataSource(coincidenceWindow), m_clientConnection(hostname, port)
 	{
+		m_eventBuilder.SetSortFlag(true);
 		m_validFlag = m_clientConnection.IsConnected();
 	}
 
@@ -12,13 +13,12 @@ namespace Specter {
 	{
 	}
 
-	const SpecData& DYOnlineSource::GetData()
+	void DYOnlineSource::ProcessData()
 	{
 		if (!m_clientConnection.IsConnected())
 		{
 			m_validFlag = false;
-			m_datum = SpecData();
-			return m_datum;
+			return;
 		}
 
 		if (m_clientConnection.GetNextEvent(m_dyHit))
@@ -28,12 +28,8 @@ namespace Specter {
 			m_datum.shortEnergy = m_dyHit.energyShort;
 			m_datum.timestamp = m_dyHit.timestamp;
 			m_datum.id = Utilities::GetBoardChannelUUID(m_dyHit.board, m_dyHit.channel);
-		}
-		else
-		{
-			m_datum = SpecData();
-		}
-
-		return m_datum;
+			if(m_eventBuilder.AddDatum(m_datum))
+				m_isEventReady = true;
+		}
 	}
 }

Specter/src/Specter/Physics/Daqromancy/DYOnlineSource.h

@@ -9,15 +9,19 @@ namespace Specter {
 	class DYOnlineSource : public DataSource
 	{
 	public:
-		DYOnlineSource(const std::string& hostname, const std::string& port);
-		~DYOnlineSource();
+		DYOnlineSource(const std::string& hostname, const std::string& port, uint64_t coincidenceWindow);
+		virtual ~DYOnlineSource();
 
-		virtual const SpecData& GetData() override;
+		virtual void ProcessData() override;
+		virtual const std::vector<SpecEvent>& GetEvents() override
+		{
+			m_isEventReady = false;
+			return m_eventBuilder.GetReadyEvents();
+		}
 
 	private:
 		DaqGrimoire::DYClient m_clientConnection;
 		DaqGrimoire::DYListData m_dyHit;
-		int m_channelsPerBoard;
 	};
 }
 

Specter/src/Specter/Physics/DataSource.cpp

@@ -11,18 +11,20 @@
 #include "Caen/CompassOnlineSource.h"
 #include "Daqromancy/DYFileSource.h"
 #include "Daqromancy/DYOnlineSource.h"
+#include "nscldaq/CharonOnlineSource.h"
 
 namespace Specter {
 
 	//loc=either an ip address or a file location, port=address port, or unused in case of file
-	DataSource* CreateDataSource(const std::string& location, const std::string& port, uint16_t header, DataSource::SourceType type)
+	DataSource* CreateDataSource(const std::string& location, const std::string& port, uint16_t header, DataSource::SourceType type, uint64_t coincidenceWindow)
 	{
 		switch(type)
 		{
-			case DataSource::SourceType::CompassOffline : return new CompassRun(location);
-			case DataSource::SourceType::CompassOnline : return new CompassOnlineSource(location, port, header);
-			case DataSource::SourceType::DaqromancyOffline: return new DYFileSource(location);
-			case DataSource::SourceType::DaqromancyOnline: return new DYOnlineSource(location, port);
+			case DataSource::SourceType::CompassOffline: return new CompassRun(location, coincidenceWindow);
+			case DataSource::SourceType::CompassOnline: return new CompassOnlineSource(location, port, header, coincidenceWindow);
+			case DataSource::SourceType::DaqromancyOffline: return new DYFileSource(location, coincidenceWindow);
+			case DataSource::SourceType::DaqromancyOnline: return new DYOnlineSource(location, port, coincidenceWindow);
+			case DataSource::SourceType::CharonOnline: return new CharonOnlineSource(location, port);
 			case DataSource::SourceType::None: return nullptr;
 		}
 		SPEC_WARN("Invalid DataSourceType at CreateDataSource!");
@@ -38,6 +40,7 @@ namespace Specter {
 			case DataSource::SourceType::CompassOffline: return "CompassOffline";
 			case DataSource::SourceType::DaqromancyOffline: return "DaqromancyOffline";
 			case DataSource::SourceType::DaqromancyOnline: return "DaqromancyOnline";
+			case DataSource::SourceType::CharonOnline: return "CharonOnline";
 		}
 
 		return "None";

Specter/src/Specter/Physics/DataSource.h

@@ -10,6 +10,7 @@
 #define DATA_SOURCE_H
 
 #include "Specter/Core/SpecCore.h"
+#include "Specter/Physics/PhysicsEventBuilder.h"
 #include "SpecData.h"
 
 namespace Specter {
@@ -23,24 +24,29 @@ namespace Specter {
 			CompassOnline,
 			CompassOffline,
 			DaqromancyOnline,
-			DaqromancyOffline
+			DaqromancyOffline,
+			CharonOnline
 		};
 
-		DataSource() :
-			m_validFlag(false)
+		DataSource(uint64_t coincidenceWindow = 0) :
+			m_validFlag(false), m_isEventReady(false), m_eventBuilder(coincidenceWindow)
 		{
 		}
 
 		virtual ~DataSource() {};
-		virtual const SpecData& GetData() = 0;
+		virtual void ProcessData() = 0;
+		virtual const std::vector<SpecEvent>& GetEvents() = 0;
 		inline bool IsValid() { return m_validFlag; }
+		inline bool IsEventReady() { return m_isEventReady; }
 
 	protected:
 		bool m_validFlag;
+		bool m_isEventReady;
 		SpecData m_datum;
+		PhysicsEventBuilder m_eventBuilder;
 	};
 
-	DataSource* CreateDataSource(const std::string& location, const std::string& port, uint16_t bitflags, DataSource::SourceType type);
+	DataSource* CreateDataSource(const std::string& location, const std::string& port, uint16_t bitflags, DataSource::SourceType type, uint64_t coincidenceWindow);
 
 	std::string ConvertDataSourceTypeToString(DataSource::SourceType type);
 }

Specter/src/Specter/Physics/PhysicsEventBuilder.cpp

@@ -12,6 +12,11 @@
 
 namespace Specter {
 
+	PhysicsEventBuilder::PhysicsEventBuilder() :
+		m_sortFlag(false), m_coincWindow(0), m_bufferIndex(0)
+	{
+	}
+
 	PhysicsEventBuilder::PhysicsEventBuilder(uint64_t windowSize) :
 		m_sortFlag(false), m_coincWindow(windowSize), m_bufferIndex(0)
 	{

Specter/src/Specter/Physics/PhysicsEventBuilder.h

@@ -18,6 +18,7 @@ namespace Specter {
 	class PhysicsEventBuilder
 	{
 	public:
+		PhysicsEventBuilder();
 		PhysicsEventBuilder(uint64_t windowSize);
 		~PhysicsEventBuilder();
 		inline void SetCoincidenceWindow(uint64_t windowSize) { m_coincWindow = windowSize; }

Specter/src/Specter/Physics/PhysicsLayer.cpp

@@ -13,7 +13,7 @@
 namespace Specter {
 
 	PhysicsLayer::PhysicsLayer(const SpectrumManager::Ref& manager) :
-		m_manager(manager), m_activeFlag(false), m_source(nullptr), m_eventBuilder(0), m_physThread(nullptr)
+		m_manager(manager), m_activeFlag(false), m_source(nullptr), m_physThread(nullptr)
 	{
 	}
 
@@ -29,12 +29,6 @@ namespace Specter {
 
 	void PhysicsLayer::OnAttach()
 	{
-		/* For debugging
-		NavParameter par("joseph");
-		par.SetValue(8);
-		NAV_INFO("Does the par exist? {0}", ParameterMap::GetInstance().IsParameterValid("joseph"));
-		NAV_INFO("What is its value? {0}", ParameterMap::GetInstance().GetParameterValue("joseph"));
-		*/
 	}
 
 	void PhysicsLayer::OnDetach()
@@ -114,10 +108,7 @@ namespace Specter {
 	{
 		SPEC_PROFILE_FUNCTION();
 		std::scoped_lock<std::mutex> guard(m_sourceMutex); //Shouldn't matter for this, but safety first
-		m_source.reset(CreateDataSource(event.GetSourceLocation(), event.GetSourcePort(), event.GetBitFlags(), event.GetSourceType()));
-		m_eventBuilder.SetCoincidenceWindow(event.GetCoincidenceWindow());
-		m_eventBuilder.SetSortFlag(event.GetSortFlag());
-		m_eventBuilder.ClearAll(); //Protect against stopping mid-event
+		m_source.reset(CreateDataSource(event.GetSourceLocation(), event.GetSourcePort(), event.GetBitFlags(), event.GetSourceType(), event.GetCoincidenceWindow()));
 		if (m_source->IsValid())
 		{
 			SPEC_INFO("Attach successful. Enabling data pull...");
@@ -145,32 +136,24 @@ namespace Specter {
 		SPEC_PROFILE_FUNCTION();
 
 		std::vector<SpecEvent> events;
-		SpecData datum;
 		while(m_activeFlag)
 		{
 			//Scope to encapsulate access to the data source
 			{
 				std::scoped_lock<std::mutex> guard(m_sourceMutex);
 				if (m_source == nullptr || !m_source->IsValid())
-				{
-					return;
-				}
-				/*
-					Looks funny, but two conditions lead to !IsValid(). Either source prev. shutdown,
-					OR we reached end of source, indicated after prev. data grab
-				*/
-				datum = m_source->GetData();
-				if(!m_source->IsValid())
 				{
 					SPEC_INFO("End of data source.");
 					return;
 				}
+				
+				m_source->ProcessData();
+				if(m_source->IsEventReady())
+				{
+					events = m_source->GetEvents();
+				}
 			}
 
-			//Pass data from source to event builder. True from AddDatum indicates events are ready
-			if (m_eventBuilder.AddDatum(datum))
-			{
-				events = m_eventBuilder.GetReadyEvents();
 			for (auto& event : events)
 			{
 				for (auto& stage : m_physStack)
@@ -181,7 +164,9 @@ namespace Specter {
 				//Invalidate all parameters to get ready for next event
 				m_manager->InvalidateParameters();
 			}
-			}
+
+			if(!events.empty())
+				events.clear();
 		}
 	}
 }

Specter/src/Specter/Physics/PhysicsLayer.h

@@ -16,7 +16,6 @@
 #include "AnalysisStack.h"
 #include "AnalysisStage.h"
 #include "DataSource.h"
-#include "PhysicsEventBuilder.h"
 #include "Specter/Core/SpectrumManager.h"
 
 #include <thread>
@@ -55,8 +54,6 @@ namespace Specter {
 		std::mutex m_sourceMutex;
 
 		std::unique_ptr<DataSource> m_source;
-		PhysicsEventBuilder m_eventBuilder;
-
 		std::thread* m_physThread;
 
 	};

Specter/src/Specter/Physics/nscldaq/CharonClient.cpp

@@ -0,0 +1,96 @@
+#include "CharonClient.h"
+
+namespace Specter {
+
+    CharonClient::CharonClient(const std::string& hostname, const std::string& port) :
+        m_socket(m_context)
+    {
+        Connect(hostname, port);
+    }
+
+    CharonClient::~CharonClient()
+    {
+        Disconnect();
+    }
+
+    bool CharonClient::GetNextEvent(std::vector<uint8_t>& event)
+    {
+        if(m_queue.IsEmpty())
+            return false;
+
+        event = m_queue.Front().body;
+        m_queue.PopFront();
+        return true;
+    }
+
+    void CharonClient::Connect(const std::string& hostname, const std::string& port)
+    {
+        try
+		{
+			asio::ip::tcp::resolver resolver(m_context);
+			auto end_points = resolver.resolve(hostname, port);
+			asio::async_connect(m_socket, end_points,
+				[this](std::error_code ec, asio::ip::tcp::endpoint endpoint)
+				{
+					if (!ec)
+					{
+						ReadHeader();
+					}
+				}
+			);
+
+			m_ioThread = std::thread([this]() { m_context.run(); });
+            SPEC_WARN("Connected CharonClient to {0}:{1}", hostname, port);
+		}
+		catch (asio::system_error& e)
+		{
+            SPEC_WARN("Unable to connect CharonClient to {0}:{1}", hostname, port);
+			return;
+		}
+    }
+
+    void CharonClient::Disconnect()
+    {
+        if (IsConnected())
+		{
+			asio::post(m_context, [this]() { m_socket.close(); });
+		}
+
+		m_context.stop();
+		if (m_ioThread.joinable())
+			m_ioThread.join();
+    }
+
+    void CharonClient::ReadHeader()
+    {
+        asio::async_read(m_socket, asio::buffer(&m_tempMessage.size, sizeof(m_tempMessage.size)),
+			[this](std::error_code ec, std::size_t size)
+			{
+				if (!ec)
+				{
+					if (m_tempMessage.size > 0)
+					{
+						m_tempMessage.body.resize(m_tempMessage.size);
+						ReadBody();
+					}
+					else
+						ReadHeader();
+				}
+			}
+		);
+    }
+
+    void CharonClient::ReadBody()
+    {
+        asio::async_read(m_socket, asio::buffer(m_tempMessage.body.data(), m_tempMessage.body.size()),
+			[this](std::error_code ec, std::size_t size)
+			{
+				if (!ec)
+				{
+                    m_queue.PushBack(m_tempMessage);
+				}
+				ReadHeader();
+			}
+		);
+    }
+}

Specter/src/Specter/Physics/nscldaq/CharonClient.h

@@ -0,0 +1,43 @@
+#ifndef CHARON_CLIENT_H
+#define CHARON_CLIENT_H
+
+#include "Specter/Utils/ThreadSafeQueue.h"
+#include <asio.hpp>
+#include <thread>
+
+namespace Specter {
+
+    struct StygianMessage
+    {
+        uint64_t size;
+        std::vector<uint8_t> body;
+    };
+
+    class CharonClient
+    {
+    public:
+        CharonClient(const std::string& hostname, const std::string& port);
+        ~CharonClient();
+
+        bool GetNextEvent(std::vector<uint8_t>& event);
+
+        void Connect(const std::string& hostname, const std::string& port);
+        void Disconnect();
+
+        const bool IsConnected() const { return m_socket.is_open(); }
+
+    private:
+        void ReadHeader();
+        void ReadBody();
+
+        asio::io_context m_context;
+        asio::ip::tcp::socket m_socket;
+        std::thread m_ioThread;
+
+        StygianMessage m_tempMessage;
+        ThreadSafeQueue<StygianMessage> m_queue;
+    };
+
+}
+
+#endif

Specter/src/Specter/Physics/nscldaq/CharonOnlineSource.cpp

@@ -0,0 +1,67 @@
+#include "CharonOnlineSource.h"
+#include "Unpackers/CaenUnpacker.h"
+#include "Unpackers/MesyTecUnpacker.h"
+
+namespace Specter {
+
+    CharonOnlineSource::CharonOnlineSource(const std::string& hostname, const std::string& port) :
+        DataSource(0), m_client(hostname, port)
+    {
+        m_validFlag = m_client.IsConnected();
+        m_readyEvents.emplace_back();
+
+        m_unpackers.push_back(std::make_shared<CaenUnpacker>());
+        m_unpackers.push_back(std::make_shared<MesyTecUnpacker>());
+    }
+
+    CharonOnlineSource::~CharonOnlineSource()
+    {
+    }
+
+    void CharonOnlineSource::ProcessData()
+    {
+        if(!m_client.IsConnected())
+        {
+            m_validFlag = false;
+            return;
+        }
+
+        if(m_client.GetNextEvent(m_rawBuffer))
+        {
+            m_event.clear();
+            UnpackRawBuffer();
+            m_isEventReady = true;
+        }
+    }
+
+    void CharonOnlineSource::UnpackRawBuffer()
+    {
+        uint32_t* iter = (uint32_t*) m_rawBuffer.data();
+        uint32_t* end = iter + m_rawBuffer.size();
+        bool wasUnpacked = false;
+
+        UnpackerResult result;
+        while(iter < end)
+        {
+            wasUnpacked = false;
+            for(auto& unpacker : m_unpackers)
+            {
+                if(unpacker->IsHeader(*iter))
+                {
+                    result = unpacker->Unpack(iter, end);
+                    iter = result.finalPosition;
+                    wasUnpacked = true;
+                    m_event.insert(m_event.end(), result.data.begin(), result.data.end());
+                    break;
+                }
+            }
+
+            if(!wasUnpacked)
+            {
+                iter++;
+            }
+        }
+
+        m_readyEvents[0] = m_event;
+    }
+}

Specter/src/Specter/Physics/nscldaq/CharonOnlineSource.h

@@ -0,0 +1,35 @@
+#ifndef CHARON_ONLINE_SOURCE
+#define CHARON_ONLINE_SOURCE
+
+#include "Specter/Physics/DataSource.h"
+#include "CharonClient.h"
+#include "Unpackers/Unpacker.h"
+
+namespace Specter {
+
+    class CharonOnlineSource : public DataSource
+    {
+    public:
+        CharonOnlineSource(const std::string& hostname, const std::string& port);
+        virtual ~CharonOnlineSource();
+
+        virtual void ProcessData() override;
+        virtual const std::vector<SpecEvent>& GetEvents() override
+        {
+            m_isEventReady = false;
+            return m_readyEvents;
+        }
+
+    private:
+        void UnpackRawBuffer();
+
+        CharonClient m_client;
+        std::vector<uint8_t> m_rawBuffer;
+        SpecEvent m_event;
+        std::vector<SpecEvent> m_readyEvents;
+        std::vector<Unpacker::Ref> m_unpackers;
+    };
+}
+
+
+#endif

Specter/src/Specter/Physics/nscldaq/Unpackers/CaenUnpacker.cpp

@@ -0,0 +1,65 @@
+#include "CaenUnpacker.h"
+#include "Specter/Utils/Functions.h"
+
+namespace Specter {
+
+    CaenUnpacker::CaenUnpacker() :
+        Unpacker()
+    {
+    }
+
+    CaenUnpacker::~CaenUnpacker() {}
+
+    void CaenUnpacker::UnpackHeader(uint32_t* word)
+    {
+        if(!IsHeader(*word))
+        {
+            SPEC_WARN("In CaenADCUnpacker::UnpackHeader() found non-header word!");
+            m_bodyWordCount = 0;
+            m_moduleID = s_illegalModuleID;
+            return;
+        }
+
+        m_bodyWordCount = (*word & s_headerCountMask) >> s_headerCountShift;
+        m_moduleID = (*word & s_geoAddressMask) >> s_geoAddressShift;
+    }
+
+    void CaenUnpacker::UnpackEnd(uint32_t* word)
+    {
+        if(!IsEnd(*word))
+        {
+            SPEC_WARN("In CaenADCUnpacker::UnpackEnd() found non-end word!");
+        }
+        //CAEN doesnt really put anything useful here
+    }
+
+    void CaenUnpacker::UnpackDatum(uint32_t* word)
+    {
+        if(!IsBody(*word))
+        {
+            SPEC_WARN("In CaenADCUnpacker::UnpackDatum() found non-body word!");
+            return;
+        }
+
+        SpecData datum;
+        uint32_t channel = (*word & s_dataChannelMask) >> s_dataChannelShift;
+        datum.id = Utilities::GetBoardChannelUUID(m_moduleID, channel);
+        datum.longEnergy = (*word & s_dataMask);
+        m_parsedEvent.data.push_back(datum);
+    }
+
+    bool CaenUnpacker::IsHeader(uint32_t word) const
+    {
+        return ((word & s_typeMask) == s_typeHeader);
+    }
+
+    bool CaenUnpacker::IsBody(uint32_t word) const
+    {
+        return ((word & s_typeMask) == s_typeBody);
+    }
+
+    bool CaenUnpacker::IsEnd(uint32_t word) const
+    {
+        return ((word & s_typeMask) == s_typeEnd);
+    }
+}

Specter/src/Specter/Physics/nscldaq/Unpackers/CaenUnpacker.h

@@ -0,0 +1,46 @@
+#ifndef CAEN_UNPACKER_H
+#define CAEN_UNPACKER_H
+
+#include "Unpacker.h"
+
+namespace Specter {
+
+    class CaenUnpacker : public Unpacker
+    {
+    public:
+        CaenUnpacker();
+        virtual ~CaenUnpacker();
+
+        virtual bool IsHeader(uint32_t word) const override;
+
+    private:
+        virtual bool IsBody(uint32_t word) const override;
+        virtual bool IsEnd(uint32_t word) const override;
+
+        virtual void UnpackHeader(uint32_t* word) override;
+        virtual void UnpackEnd(uint32_t* word) override;
+
+        virtual void UnpackDatum(uint32_t* word) override;
+
+        static constexpr uint32_t s_typeMask = 0x07000000;
+        static constexpr uint32_t s_typeHeader = 0x02000000;
+        static constexpr uint32_t s_typeBody = 0x00000000;
+        static constexpr uint32_t s_typeEnd = 0x04000000;
+
+        static constexpr uint32_t s_geoAddressShift = 27;
+        static constexpr uint32_t s_geoAddressMask = 0xf8000000;
+
+        static constexpr uint32_t s_headerCountShift = 8;
+        static constexpr uint32_t s_headerCountMask = 0x00003f00;
+
+        static constexpr uint32_t s_dataChannelShift = 16;
+        static constexpr uint32_t s_dataChannelMask = 0x001f0000;
+        static constexpr uint32_t s_dataMask = 0x00000fff;
+        static constexpr uint32_t s_dataStatusMask = 0x00003000;
+        static constexpr uint32_t s_dataOverflow = 0x00001000;
+        static constexpr uint32_t s_dataUnderflow = 0x00002000;
+    };
+}
+
+
+#endif

Specter/src/Specter/Physics/nscldaq/Unpackers/MesyTecUnpacker.cpp

@@ -0,0 +1,62 @@
+#include "MesyTecUnpacker.h"
+#include "Specter/Utils/Functions.h"
+
+namespace Specter {
+
+    MesyTecUnpacker::MesyTecUnpacker() :
+        Unpacker()
+    {
+    }
+
+    MesyTecUnpacker::~MesyTecUnpacker() {}
+
+    void MesyTecUnpacker::UnpackHeader(uint32_t* word)
+    {
+        if(!IsHeader(*word))
+        {
+            SPEC_WARN("In MesyQDCUnpacker::UnpackHeader() found non-header word!");
+            m_moduleID = s_illegalModuleID;
+            m_bodyWordCount = 0;
+        }
+
+        m_moduleID = (*word & s_idMask) >> s_idShift;
+        m_bodyWordCount = (*word & s_headerCountMask) - 1; //For MesyTec, count includes the end word
+    }
+
+    void MesyTecUnpacker::UnpackDatum(uint32_t* word)
+    {
+        if(!IsBody(*word))
+        {
+            SPEC_WARN("In MesyQDCUnpacker::UnpackDatum() found non-body word!");
+            return;
+        }
+        
+        SpecData datum;
+        uint32_t channel = (*word & s_dataChannelMask) >> s_dataChannelShift;
+        datum.id = Utilities::GetBoardChannelUUID(m_moduleID, channel);
+        datum.longEnergy = (*word & s_dataMask);
+    }
+
+    void MesyTecUnpacker::UnpackEnd(uint32_t* word)
+    {
+        if(!IsEnd(*word))
+        {
+            SPEC_WARN("In MesyQDCUnpacker::UnpackEnd() found non-end word!");
+        }
+    }
+
+    bool MesyTecUnpacker::IsHeader(uint32_t word) const
+    {
+        return ((word & s_typeMask) == s_typeHeader);
+    }
+
+    bool MesyTecUnpacker::IsBody(uint32_t word) const
+    {
+        return ((word & s_typeMask) == s_typeBody);
+    }
+
+    bool MesyTecUnpacker::IsEnd(uint32_t word) const
+    {
+        return ((word & s_typeMask) == s_typeEnd);
+    }
+}

Specter/src/Specter/Physics/nscldaq/Unpackers/MesyTecUnpacker.h

@@ -0,0 +1,41 @@
+#ifndef MESY_TEC_UNPACKER_H
+#define MESY_TEC_UNPACKER_H
+
+#include "Unpacker.h"
+
+namespace Specter {
+
+    class MesyTecUnpacker : public Unpacker
+    {
+    public:
+        MesyTecUnpacker();
+        virtual ~MesyTecUnpacker();
+
+        virtual bool IsHeader(uint32_t word) const override;
+
+    private:
+        virtual bool IsBody(uint32_t word) const override;
+        virtual bool IsEnd(uint32_t word) const override;
+
+        virtual void UnpackHeader(uint32_t* word) override;
+        virtual void UnpackEnd(uint32_t* word) override;
+
+        virtual void UnpackDatum(uint32_t* word) override;
+
+        static constexpr uint32_t s_typeMask = 0xc000000;
+        static constexpr uint32_t s_typeHeader = 0x40000000;
+        static constexpr uint32_t s_typeBody = 0x00000000;
+        static constexpr uint32_t s_typeEnd = 0xc0000000;
+
+        static constexpr uint32_t s_idShift = 16;
+        static constexpr uint32_t s_idMask = 0x00ff0000;
+
+        static constexpr uint32_t s_headerCountMask = 0x00000fff;
+
+        static constexpr uint32_t s_dataChannelShift = 16;
+        static constexpr uint32_t s_dataChannelMask = 0x001f0000;
+        static constexpr uint32_t s_dataMask = 0x0000ffff;
+    };
+}
+
+#endif

Specter/src/Specter/Physics/nscldaq/Unpackers/Unpacker.h

@@ -0,0 +1,82 @@
+#ifndef UNPACKER_H
+#define UNPACKER_H
+
+#include "Specter/Physics/SpecData.h"
+
+namespace Specter {
+
+    struct UnpackerResult
+    {
+        uint32_t* finalPosition = nullptr;
+        std::vector<SpecData> data;
+    };
+
+    class Unpacker
+    {
+    public:
+        using Ref = std::shared_ptr<Unpacker>;
+
+        Unpacker() :
+            m_bodyWordCount(0), m_moduleID(s_illegalModuleID)
+        {
+
+        }
+        virtual ~Unpacker() {}
+
+        const UnpackerResult& Unpack(uint32_t* begin, uint32_t* end)
+        {
+            m_parsedEvent = UnpackerResult();
+
+            uint32_t* iter = begin;
+            UnpackHeader(iter);
+            ++iter;
+
+            uint32_t* bodyEnd = iter + m_bodyWordCount;
+            if(bodyEnd > end || m_moduleID == s_illegalModuleID)
+            {
+                SPEC_WARN("In Unpacker::Unpack() header unpack error (number of words: {0}, moduleID: {1}), data not parsed!", m_bodyWordCount, m_moduleID);
+            }
+            else
+            {
+                iter = UnpackBody(iter, bodyEnd);
+                UnpackEnd(iter);
+                ++iter;
+            }
+
+            m_parsedEvent.finalPosition = iter;
+
+            return m_parsedEvent;
+        }
+
+        virtual bool IsHeader(uint32_t word) const = 0;
+
+    protected:
+        virtual bool IsBody(uint32_t word) const = 0;
+        virtual bool IsEnd(uint32_t word) const = 0;
+
+        virtual void UnpackHeader(uint32_t* word) = 0;
+        virtual void UnpackEnd(uint32_t* word) = 0;
+
+        uint32_t* UnpackBody(uint32_t* bodyBegin, uint32_t* bodyEnd)
+        {
+            uint32_t* iter = bodyBegin;
+            while(iter != bodyEnd)
+            {
+                UnpackDatum(iter);
+                iter++;
+            }
+
+            return iter;
+        }
+
+        virtual void UnpackDatum(uint32_t* word) = 0;
+
+        UnpackerResult m_parsedEvent;
+        uint64_t m_bodyWordCount;
+        uint32_t m_moduleID;
+
+        static constexpr uint32_t s_illegalModuleID = 999;
+    };
+}
+
+#endif

Specter/src/Specter/Utils/ThreadSafeQueue.h

@@ -0,0 +1,128 @@
+#ifndef SPECTER_THREAD_SAFE_QUEUE_H
+#define SPECTER_THREAD_SAFE_QUEUE_H
+
+#include <thread>
+#include <mutex>
+#include <atomic>
+#include <queue>
+#include <condition_variable>
+
+namespace Specter {
+
+    template<typename T>
+	class ThreadSafeQueue
+	{
+	public:
+		ThreadSafeQueue() = default;
+		ThreadSafeQueue(const ThreadSafeQueue&) = delete; //no copy
+		~ThreadSafeQueue() { Clear(); }
+
+		void PushBack(const T& data)
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			m_queue.push_back(data);
+
+			std::scoped_lock<std::mutex> condGuard(m_conditionMutex);
+			m_conditional.notify_one();
+		}
+
+		void PushFront(const T& data)
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			m_queue.push_front(data);
+
+			std::scoped_lock<std::mutex> conditionGuard(m_conditionMutex);
+			m_conditional.notify_one();
+		}
+
+		void PopBack()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			m_queue.pop_back();
+		}
+
+		void PopFront()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			m_queue.pop_front();
+		}
+
+		const T& Front()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			return m_queue.front();
+		}
+
+		const T& Back()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			return m_queue.back();
+		}
+
+		std::size_t Size()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			return m_queue.size();
+		}
+
+		bool IsEmpty()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			return m_queue.empty();
+		}
+
+		void Clear()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			m_queue.clear();
+		}
+
+		//For iterator loops, need begin()/end() idiom
+
+		typename std::deque<T>::iterator begin()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			return m_queue.begin();
+		}
+
+		typename std::deque<T>::iterator end()
+		{
+			std::scoped_lock<std::mutex> guard(m_queueMutex);
+			return m_queue.end();
+		}
+
+		void Wait()
+		{
+			while (IsEmpty() && !m_isForceWakeup)
+			{
+				std::unique_lock<std::mutex> guard(m_conditionMutex);
+				m_conditional.wait(guard);
+			}
+		}
+
+		void ForceWakeup()
+		{
+			m_isForceWakeup = true;
+
+			std::unique_lock<std::mutex> guard(m_conditionMutex);
+			m_conditional.notify_one();
+		}
+
+		void ResetWakeup()
+		{
+			m_isForceWakeup = false;
+		}
+
+	private:
+		std::mutex m_queueMutex;
+		std::deque<T> m_queue;
+
+		std::mutex m_conditionMutex;
+		std::condition_variable m_conditional;
+
+		std::atomic<bool> m_isForceWakeup = false;
+	};
+    
+}
+
+#endif