#include <iostream>
#include <thread>
#include <cstring>
#include <sys/socket.h>
#include <unistd.h>
#include <netinet/in.h>
#include <mutex>
#include <condition_variable>
// Solving Producer consumer problem using Message Passing
// (socket programming process to process communication)
std::mutex mtx;
std::condition_variable cv;
bool isProducerReady = false;
#define N 100 // on 100 message can be resolved at a time
// Producer code
class ProducerClass
{
private:
int item=0;
public:
void operator()(void)
{
// Server code(producer) here which will send msg on demand to client (consumer)
// creating socket, server socket is a file descriptor
int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
if (serverSocket == -1)
{
perror("Socket creation failed");
return;
}
// Allow reusing the same address to prevent binding failures on quick restarts
int opt = 1;
setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
// defining the address
sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_port = htons(8080);
serverAddress.sin_addr.s_addr = INADDR_ANY;
// bind socket
if (bind(serverSocket, (struct sockaddr *)&serverAddress, sizeof(serverAddress)) < 0)
{
perror("Bind failed");
close(serverSocket);
return;
}
// listen to assigned socket
if (listen(serverSocket, 2) < 0)
{
perror("Listen failed");
close(serverSocket);
return;
}
// when producer is ready i.e true, release the lock
{
std::lock_guard<std::mutex> lock(mtx);
isProducerReady = true;
cv.notify_one();
}
// accept connections
int clientSocket = accept(serverSocket, nullptr, nullptr);
if (clientSocket < 0)
{
perror("Accept failed");
close(serverSocket);
return;
}
char buffer[1024] = {0};
std::string leftover = "";
while (true)
{
int bytesReceived = recv(clientSocket, buffer, sizeof(buffer) - 1, 0);
if (bytesReceived <= 0) break; // Connection closed
buffer[bytesReceived] = '\0'; // null termination
std::string receivedData = leftover + std::string(buffer);
leftover.clear();
size_t pos;
while ((pos = receivedData.find('\n')) != std::string::npos) {
std::string message = receivedData.substr(0, pos);
receivedData.erase(0, pos + 1);
if (!message.empty()) {
std::cout << "\nReceived: " << message << std::endl;
{
std::lock_guard<std::mutex> lock(mtx);
item++;
}
std::string response = std::to_string(item) + "\n";
send(clientSocket, response.c_str(), response.length(), 0);
}
}
leftover = receivedData;
}
close(clientSocket);
close(serverSocket);
}
};
// Consumer code
class ConsumerClass
{
public:
void operator()(void)
{
std::unique_lock<std::mutex> lock(mtx);
// wait until producer is ready(server setups)
cv.wait(lock, [] { return isProducerReady; });
lock.unlock();
// creating socket
int clientSocket = socket(AF_INET, SOCK_STREAM, 0);
if (clientSocket == -1)
{
perror("Socket creation failed");
return;
}
// specifying address
sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_port = htons(8080);
serverAddress.sin_addr.s_addr = INADDR_ANY;
// sending connection request
if (connect(clientSocket, (struct sockaddr *)&serverAddress, sizeof(serverAddress)) < 0)
{
perror("Connection failed");
close(clientSocket);
return;
}
// send empty messages
for(int i=1;i<=N;i++){
// sending data
std::string message = "_empty_string_\n";
send(clientSocket, message.c_str(), message.length(), 0);
}
// Receive responses
char buffer[1024] = {0};
int totalResponses = 0;
while (totalResponses < N) {
int bytesRead = recv(clientSocket, buffer, sizeof(buffer) - 1, 0);
if (bytesRead <= 0) break; // Stop when the server closes connection
buffer[bytesRead] = '\0';
std::string receivedData(buffer);
size_t pos;
while ((pos = receivedData.find('\n')) != std::string::npos) {
std::string response = receivedData.substr(0, pos);
receivedData.erase(0, pos + 1);
if (!response.empty()) {
std::cout << "Server Response: " << response << std::endl;
totalResponses++;
}
}
}
// Close socket
close(clientSocket);
}
};
int main()
{
ProducerClass producer;
ConsumerClass consumer;
// create two threads for producer and client
std::thread t1(producer);
std::thread t2(consumer);
t1.join();
t2.join();
return 0;
}
