diff --git a/src/servers/media/Jamfile b/src/servers/media/Jamfile index b9b571d91d..2114581b42 100644 --- a/src/servers/media/Jamfile +++ b/src/servers/media/Jamfile @@ -7,6 +7,7 @@ Server media_server : AppManager.cpp BufferManager.cpp NodeManager.cpp + NotificationProcessor.cpp Queue.cpp ; LinkSharedOSLibs media_server : be libmedia.so root ; diff --git a/src/servers/media/NotificationProcessor.cpp b/src/servers/media/NotificationProcessor.cpp new file mode 100644 index 0000000000..01967de052 --- /dev/null +++ b/src/servers/media/NotificationProcessor.cpp @@ -0,0 +1,67 @@ +#include +#include +#include "NotificationProcessor.h" +#include "Queue.h" + +#define NOTIFICATION_THREAD_PRIORITY 19 + +NotificationProcessor::NotificationProcessor() + : fNotificationQueue(new Queue), + fNotificationThreadId(-1) +{ + fNotificationThreadId = spawn_thread(NotificationProcessor::worker_thread, "notification broadcast", NOTIFICATION_THREAD_PRIORITY, this); + resume_thread(fNotificationThreadId); +} + +NotificationProcessor::~NotificationProcessor() +{ + // properly terminate the queue and wait until the worker thread has finished + status_t dummy; + fNotificationQueue->Terminate(); + wait_for_thread(fNotificationThreadId, &dummy); + delete fNotificationQueue; +} + +void +NotificationProcessor::RequestNotifications(BMessage *msg) +{ +} + +void +NotificationProcessor::CancelNotifications(BMessage *msg) +{ +} + +void +NotificationProcessor::SendNotifications(BMessage *msg) +{ + // queue a copy of the message to be processed later + fNotificationQueue->AddItem(new BMessage(*msg)); +} + +void +NotificationProcessor::CleanupTeam(team_id team) +{ +} + +void +NotificationProcessor::BroadcastMessages(BMessage *msg) +{ +} + +void +NotificationProcessor::WorkerThread() +{ + BMessage *msg; + while (NULL != (msg = static_cast(fNotificationQueue->RemoveItem()))) { + BroadcastMessages(msg); + delete msg; + } +} + +int32 +NotificationProcessor::worker_thread(void *arg) +{ + static_cast(arg)->WorkerThread(); + return 0; +} diff --git a/src/servers/media/NotificationProcessor.h b/src/servers/media/NotificationProcessor.h new file mode 100644 index 0000000000..61e4786367 --- /dev/null +++ b/src/servers/media/NotificationProcessor.h @@ -0,0 +1,24 @@ + +class Queue; + +class NotificationProcessor +{ +public: + NotificationProcessor(); + ~NotificationProcessor(); + + void RequestNotifications(BMessage *msg); + void CancelNotifications(BMessage *msg); + void SendNotifications(BMessage *msg); + + void CleanupTeam(team_id team); + +private: + void BroadcastMessages(BMessage *msg); + void WorkerThread(); + static int32 worker_thread(void *arg); + +private: + Queue * fNotificationQueue; + thread_id fNotificationThreadId; +}; diff --git a/src/servers/media/media_server.cpp b/src/servers/media/media_server.cpp index 204fc7cab1..e1a3cf4aac 100644 --- a/src/servers/media/media_server.cpp +++ b/src/servers/media/media_server.cpp @@ -4,6 +4,7 @@ #include #include #include +#include "NotificationProcessor.h" #include "ServerInterface.h" #include "BufferManager.h" #include "NodeManager.h" @@ -64,8 +65,6 @@ public: void UnregisterNode(BMessage *); void SetDefault(BMessage *); void AcquireNodeReference(BMessage *); - void RequestNotifications(BMessage *); - void CancelNotifications(BMessage *); void SetOutputBuffers(BMessage *); void ReclaimOutputBuffers(BMessage *); void OrphanReclaimableBuffers(BMessage *); @@ -107,6 +106,7 @@ private: port_id control_port; thread_id control_thread; + NotificationProcessor *fNotificationProcessor; BufferManager *fBufferManager; AppManager *fAppManager; NodeManager *fNodeManager; @@ -121,6 +121,7 @@ private: ServerApp::ServerApp() : BApplication(NEW_MEDIA_SERVER_SIGNATURE), + fNotificationProcessor(new NotificationProcessor), fBufferManager(new BufferManager), fAppManager(new AppManager), fNodeManager(new NodeManager), @@ -138,6 +139,7 @@ ServerApp::ServerApp() ServerApp::~ServerApp() { + delete fNotificationProcessor; delete fBufferManager; delete fAppManager; delete fNodeManager; @@ -447,16 +449,6 @@ void ServerApp::AcquireNodeReference(BMessage *msg) } -void ServerApp::RequestNotifications(BMessage *msg) -{ -} - - -void ServerApp::CancelNotifications(BMessage *msg) -{ -} - - void ServerApp::SetOutputBuffers(BMessage *msg) { } @@ -555,6 +547,9 @@ void ServerApp::MessageReceived(BMessage *msg) case MEDIA_SERVER_GET_SHARED_BUFFER_AREA: GetSharedBufferArea(msg); break; case MEDIA_SERVER_REGISTER_BUFFER: RegisterBuffer(msg); break; case MEDIA_SERVER_UNREGISTER_BUFFER: UnregisterBuffer(msg); break; + case MEDIA_SERVER_REQUEST_NOTIFICATIONS: fNotificationProcessor->RequestNotifications(msg); break; + case MEDIA_SERVER_CANCEL_NOTIFICATIONS: fNotificationProcessor->CancelNotifications(msg); break; + case MEDIA_SERVER_SEND_NOTIFICATIONS: fNotificationProcessor->SendNotifications(msg); break; case MEDIA_SERVER_GET_NODE_ID: GetNodeID(msg); break; @@ -576,8 +571,6 @@ void ServerApp::MessageReceived(BMessage *msg) case MEDIA_SERVER_UNREGISTER_NODE: UnregisterNode(msg); break; case MEDIA_SERVER_SET_DEFAULT: SetDefault(msg); break; case MEDIA_SERVER_ACQUIRE_NODE_REFERENCE: AcquireNodeReference(msg); break; - case MEDIA_SERVER_REQUEST_NOTIFICATIONS: RequestNotifications(msg); break; - case MEDIA_SERVER_CANCEL_NOTIFICATIONS: CancelNotifications(msg); break; case MEDIA_SERVER_SET_OUTPUT_BUFFERS: SetOutputBuffers(msg); break; case MEDIA_SERVER_RECLAIM_OUTPUT_BUFFERS: ReclaimOutputBuffers(msg); break; case MEDIA_SERVER_ORPHAN_RECLAIMABLE_BUFFERS: OrphanReclaimableBuffers(msg); break;