diff --git a/src/CvSourceImpl.cpp b/src/CvSourceImpl.cpp
index bd02d2aebf..8fcbb3f4b7 100644
--- a/src/CvSourceImpl.cpp
+++ b/src/CvSourceImpl.cpp
@@ -37,8 +37,6 @@ void CvSourceImpl::Start() {
       CreateProperty("jpeg_quality", CS_PROP_INTEGER, 0, 100, 1, 80, 80);
 }
 
-bool CvSourceImpl::IsConnected() const { return m_connected; }
-
 bool CvSourceImpl::CacheProperties(CS_Status* status) const {
   // Doesn't need to do anything.
   m_properties_cached = true;
@@ -112,14 +110,6 @@ void CvSourceImpl::NotifyError(llvm::StringRef msg) {
   PutError(msg, wpi::Now());
 }
 
-void CvSourceImpl::SetConnected(bool connected) {
-  bool was_connected = m_connected.exchange(connected);
-  if (was_connected && !connected)
-    Notifier::GetInstance().NotifySource(*this, CS_SOURCE_DISCONNECTED);
-  else if (!was_connected && connected)
-    Notifier::GetInstance().NotifySource(*this, CS_SOURCE_CONNECTED);
-}
-
 int CvSourceImpl::CreateProperty(llvm::StringRef name, CS_PropertyKind kind,
                                  int minimum, int maximum, int step,
                                  int defaultValue, int value) {
diff --git a/src/CvSourceImpl.h b/src/CvSourceImpl.h
index 4fd88cd478..b05a2fb07b 100644
--- a/src/CvSourceImpl.h
+++ b/src/CvSourceImpl.h
@@ -23,8 +23,6 @@ class CvSourceImpl : public SourceImpl {
 
   void Start();
 
-  bool IsConnected() const override;
-
   // Property functions
   void SetProperty(int property, int value, CS_Status* status) override;
   void SetStringProperty(int property, llvm::StringRef value,
@@ -38,7 +36,6 @@ class CvSourceImpl : public SourceImpl {
   // OpenCV-specific functions
   void PutFrame(cv::Mat& image);
   void NotifyError(llvm::StringRef msg);
-  void SetConnected(bool connected);
   int CreateProperty(llvm::StringRef name, CS_PropertyKind kind, int minimum,
                      int maximum, int step, int defaultValue, int value);
   int CreateProperty(llvm::StringRef name, CS_PropertyKind kind, int minimum,
diff --git a/src/SourceImpl.cpp b/src/SourceImpl.cpp
index c259bfa84d..d3ad6a9613 100644
--- a/src/SourceImpl.cpp
+++ b/src/SourceImpl.cpp
@@ -11,6 +11,7 @@
 #include <cstring>
 
 #include "Log.h"
+#include "Notifier.h"
 
 using namespace cs;
 
@@ -44,6 +45,14 @@ llvm::StringRef SourceImpl::GetDescription(
   return llvm::StringRef{buf.data(), buf.size()};
 }
 
+void SourceImpl::SetConnected(bool connected) {
+  bool wasConnected = m_connected.exchange(connected);
+  if (wasConnected && !connected)
+    Notifier::GetInstance().NotifySource(*this, CS_SOURCE_DISCONNECTED);
+  else if (!wasConnected && connected)
+    Notifier::GetInstance().NotifySource(*this, CS_SOURCE_CONNECTED);
+}
+
 uint64_t SourceImpl::GetCurFrameTime() {
   std::unique_lock<std::mutex> lock{m_frameMutex};
   return m_frame.time();
diff --git a/src/SourceImpl.h b/src/SourceImpl.h
index c72ba94ce6..9a769d15fb 100644
--- a/src/SourceImpl.h
+++ b/src/SourceImpl.h
@@ -8,6 +8,7 @@
 #ifndef CS_SOURCEIMPL_H_
 #define CS_SOURCEIMPL_H_
 
+#include <atomic>
 #include <condition_variable>
 #include <cstddef>
 #include <memory>
@@ -37,7 +38,8 @@ class SourceImpl {
   void SetDescription(llvm::StringRef description);
   llvm::StringRef GetDescription(llvm::SmallVectorImpl<char>& buf) const;
 
-  virtual bool IsConnected() const = 0;
+  void SetConnected(bool connected);
+  bool IsConnected() const { return m_connected; }
 
   // Functions to keep track of the overall number of sinks connected to this
   // source.  Primarily used by sinks to determine if other sinks are using
@@ -210,6 +212,8 @@ class SourceImpl {
   // Pool of frame data to reduce malloc traffic.
   std::mutex m_poolMutex;
   std::vector<std::unique_ptr<Frame::Data>> m_framesAvail;
+
+  std::atomic_bool m_connected{false};
 };
 
 }  // namespace cs
diff --git a/src/UsbCameraImpl.cpp b/src/UsbCameraImpl.cpp
index 10c26ec7ba..cae8bd13e0 100644
--- a/src/UsbCameraImpl.cpp
+++ b/src/UsbCameraImpl.cpp
@@ -595,7 +595,7 @@ void UsbCameraImpl::DeviceDisconnect() {
   close(fd);
 
   // Notify
-  Notifier::GetInstance().NotifySource(*this, CS_SOURCE_DISCONNECTED);
+  SetConnected(false);
 }
 
 void UsbCameraImpl::DeviceConnect() {
@@ -691,7 +691,7 @@ void UsbCameraImpl::DeviceConnect() {
   SetDescription(GetDescriptionImpl(m_path.c_str()));
 
   // Notify
-  Notifier::GetInstance().NotifySource(*this, CS_SOURCE_CONNECTED);
+  SetConnected(true);
 }
 
 bool UsbCameraImpl::DeviceStreamOn() {
@@ -1297,8 +1297,6 @@ bool UsbCameraImpl::CacheProperties(CS_Status* status) const {
   return true;
 }
 
-bool UsbCameraImpl::IsConnected() const { return m_fd >= 0; }
-
 void UsbCameraImpl::SetProperty(int property, int value, CS_Status* status) {
   auto msg = CreateMessage(Message::kCmdSetProperty);
   msg->data[0] = property;
diff --git a/src/UsbCameraImpl.h b/src/UsbCameraImpl.h
index 8f27d8710e..da44196573 100644
--- a/src/UsbCameraImpl.h
+++ b/src/UsbCameraImpl.h
@@ -33,8 +33,6 @@ class UsbCameraImpl : public SourceImpl {
 
   void Start();
 
-  bool IsConnected() const override;
-
   // Property functions
   void SetProperty(int property, int value, CS_Status* status) override;
   void SetStringProperty(int property, llvm::StringRef value,