/*
 * Compact EPICS Channel Access server for scalar, pointer-backed values.
 *
 * It implements only discovery, channel creation, scalar get/put and change
 * subscriptions, access rights, echo, and channel cleanup. Arrays, beacons,
 * records, and automatic value locking are intentionally outside this server.
 *
 * Generated with OpenAI Codex 5.6 Sol-high, 29.08.2026, S. Ritt
 */

#include "epics_ioc.h"

#include "epics_ca.h"

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include <algorithm>
#include <cmath>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <limits>
#include <memory>
#include <new>
#include <string>
#include <unordered_map>
#include <vector>

namespace midas {
namespace epics {
namespace {

const std::uint16_t kProtocolRevision = 13;
const std::size_t kHeaderSize = 16;
const std::size_t kMaxPayloadSize = 1024 * 1024;
const std::uint32_t kEpicsEpochOffset = 631152000;

enum Command : std::uint16_t {
   kVersion = 0,
   kEventAdd = 1,
   kEventCancel = 2,
   kWrite = 4,
   kSearch = 6,
   kClearChannel = 12,
   kReadNotify = 15,
   kCreateChannel = 18,
   kWriteNotify = 19,
   kClientName = 20,
   kHostName = 21,
   kAccessRights = 22,
   kEcho = 23
};

struct Header {
   std::uint16_t command;
   std::uint32_t payload_size;
   std::uint16_t data_type;
   std::uint32_t data_count;
   std::uint32_t parameter1;
   std::uint32_t parameter2;
   std::size_t header_size;
};

std::uint16_t read_u16(const std::uint8_t *data)
{
   std::uint16_t value;
   std::memcpy(&value, data, sizeof(value));
   return ntohs(value);
}

std::uint32_t read_u32(const std::uint8_t *data)
{
   std::uint32_t value;
   std::memcpy(&value, data, sizeof(value));
   return ntohl(value);
}

std::uint64_t read_u64(const std::uint8_t *data)
{
   return (static_cast<std::uint64_t>(read_u32(data)) << 32) |
          read_u32(data + 4);
}

void append_u16(std::vector<std::uint8_t> &buffer, std::uint16_t value)
{
   value = htons(value);
   const std::uint8_t *bytes = reinterpret_cast<const std::uint8_t *>(&value);
   buffer.insert(buffer.end(), bytes, bytes + sizeof(value));
}

void append_u32(std::vector<std::uint8_t> &buffer, std::uint32_t value)
{
   value = htonl(value);
   const std::uint8_t *bytes = reinterpret_cast<const std::uint8_t *>(&value);
   buffer.insert(buffer.end(), bytes, bytes + sizeof(value));
}

void append_u64(std::vector<std::uint8_t> &buffer, std::uint64_t value)
{
   append_u32(buffer, static_cast<std::uint32_t>(value >> 32));
   append_u32(buffer, static_cast<std::uint32_t>(value));
}

std::size_t aligned_size(std::size_t size)
{
   return (size + 7) & ~static_cast<std::size_t>(7);
}

void append_message(std::vector<std::uint8_t> &buffer, std::uint16_t command,
                    std::uint16_t data_type, std::uint32_t data_count,
                    std::uint32_t parameter1, std::uint32_t parameter2,
                    const std::uint8_t *payload = NULL,
                    std::size_t payload_size = 0)
{
   const std::size_t padded = aligned_size(payload_size);
   if (padded > std::numeric_limits<std::uint16_t>::max())
      return;
   append_u16(buffer, command);
   append_u16(buffer, static_cast<std::uint16_t>(padded));
   append_u16(buffer, data_type);
   append_u16(buffer, static_cast<std::uint16_t>(data_count));
   append_u32(buffer, parameter1);
   append_u32(buffer, parameter2);
   if (payload_size)
      buffer.insert(buffer.end(), payload, payload + payload_size);
   buffer.insert(buffer.end(), padded - payload_size, 0);
}

bool decode_header(const std::uint8_t *data, std::size_t size, Header &header)
{
   if (size < kHeaderSize)
      return false;
   header.command = read_u16(data);
   header.payload_size = read_u16(data + 2);
   header.data_type = read_u16(data + 4);
   header.data_count = read_u16(data + 6);
   header.parameter1 = read_u32(data + 8);
   header.parameter2 = read_u32(data + 12);
   header.header_size = kHeaderSize;
   if (header.payload_size == 0xffff && header.data_count == 0) {
      if (size < 24)
         return false;
      header.payload_size = read_u32(data + 16);
      header.data_count = read_u32(data + 20);
      header.header_size = 24;
   }
   return true;
}

bool set_nonblocking(int fd)
{
   int flags = fcntl(fd, F_GETFL, 0);
   return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0;
}

std::string socket_error(const char *operation)
{
   return std::string(operation) + ": " + std::strerror(errno);
}

void append_float(std::vector<std::uint8_t> &buffer, float value)
{
   std::uint32_t bits;
   std::memcpy(&bits, &value, sizeof(bits));
   append_u32(buffer, bits);
}

void append_double(std::vector<std::uint8_t> &buffer, double value)
{
   std::uint64_t bits;
   std::memcpy(&bits, &value, sizeof(bits));
   append_u64(buffer, bits);
}

std::string value_string(double value)
{
   char text[40];
   std::snprintf(text, sizeof(text), "%.15g", value);
   return text;
}

std::string value_string(std::int32_t value)
{
   char text[40];
   std::snprintf(text, sizeof(text), "%d", static_cast<int>(value));
   return text;
}

} // namespace

struct SoftIoc::Impl {
   enum ValueType { DoubleValue, Int32Value };

   struct Pv {
      std::string name;
      ValueType type;
      void *value;
      bool writable;
   };

   struct Channel {
      Pv *pv;
      std::uint32_t cid;
      std::uint32_t sid;
   };

   struct Monitor {
      Pv *pv;
      std::uint32_t sid;
      std::uint32_t subscription_id;
      std::uint16_t data_type;
      double last_value;
   };

   struct Client {
      int fd;
      bool dead;
      bool eof;
      std::vector<std::uint8_t> input;
      std::vector<std::uint8_t> output;
      std::size_t output_offset;
      std::unordered_map<std::uint32_t, Channel> channels;
      std::unordered_map<std::uint32_t, Monitor> monitors;

      explicit Client(int socket_fd)
         : fd(socket_fd), dead(false), eof(false), output_offset(0)
      {
      }
   };

   std::vector<std::unique_ptr<Pv> > pvs;
   std::unordered_map<std::string, Pv *> pvs_by_name;
   std::vector<std::unique_ptr<Client> > clients;
   int udp_fd = -1;
   int listen_fd = -1;
   std::uint16_t port = 0;
   std::uint32_t next_sid = 1;
   std::string error;

   ~Impl()
   {
      for (std::size_t i = 0; i < clients.size(); ++i)
         close(clients[i]->fd);
      if (listen_fd >= 0)
         close(listen_fd);
      if (udp_fd >= 0)
         close(udp_fd);
   }

   bool add(const std::string &name, void *value, ValueType type, bool writable)
   {
      if (udp_fd >= 0 || listen_fd >= 0) {
         error = "PVs cannot be added after the IOC has started";
         return false;
      }
      if (name.empty() || value == NULL || name.size() >= 512) {
         error = "PV name and value pointer must be valid";
         return false;
      }
      if (pvs_by_name.find(name) != pvs_by_name.end()) {
         error = "duplicate PV: " + name;
         return false;
      }
      std::unique_ptr<Pv> pv(new (std::nothrow) Pv);
      if (!pv.get()) {
         error = "cannot allocate PV";
         return false;
      }
      pv->name = name;
      pv->type = type;
      pv->value = value;
      pv->writable = writable;
      pvs_by_name[name] = pv.get();
      pvs.push_back(std::move(pv));
      return true;
   }

   bool start(std::uint16_t requested_port)
   {
      if (udp_fd >= 0 || listen_fd >= 0) {
         error = "IOC is already running";
         return false;
      }
      if (!requested_port) {
         error = "IOC port must be between 1 and 65535";
         return false;
      }

      udp_fd = socket(AF_INET, SOCK_DGRAM, 0);
      listen_fd = socket(AF_INET, SOCK_STREAM, 0);
      if (udp_fd < 0 || listen_fd < 0) {
         error = socket_error("socket");
         return false;
      }

      int enabled = 1;
      setsockopt(udp_fd, SOL_SOCKET, SO_REUSEADDR, &enabled, sizeof(enabled));
      setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enabled, sizeof(enabled));

      sockaddr_in address;
      std::memset(&address, 0, sizeof(address));
      address.sin_family = AF_INET;
      address.sin_addr.s_addr = htonl(INADDR_ANY);
      address.sin_port = htons(requested_port);
      if (bind(udp_fd, reinterpret_cast<sockaddr *>(&address), sizeof(address)) < 0) {
         error = socket_error("UDP bind");
         return false;
      }
      if (bind(listen_fd, reinterpret_cast<sockaddr *>(&address), sizeof(address)) < 0 ||
          listen(listen_fd, 16) < 0) {
         error = socket_error("TCP bind/listen");
         return false;
      }
      if (!set_nonblocking(udp_fd) || !set_nonblocking(listen_fd)) {
         error = socket_error("nonblocking socket setup");
         return false;
      }
      port = requested_port;
      error.clear();
      return true;
   }

   void receive_discovery()
   {
      for (;;) {
         std::uint8_t data[65536];
         sockaddr_in source;
         socklen_t source_size = sizeof(source);
         ssize_t received = recvfrom(udp_fd, data, sizeof(data), 0,
            reinterpret_cast<sockaddr *>(&source), &source_size);
         if (received < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
               return;
            return;
         }

         std::vector<std::uint8_t> reply;
         std::size_t offset = 0;
         while (offset + kHeaderSize <= static_cast<std::size_t>(received)) {
            Header header;
            if (!decode_header(data + offset,
                               static_cast<std::size_t>(received) - offset,
                               header))
               break;
            const std::size_t total = header.header_size + header.payload_size;
            if (total > static_cast<std::size_t>(received) - offset)
               break;
            const std::uint8_t *payload = data + offset + header.header_size;
            if (header.command == kSearch && header.payload_size) {
               const std::size_t length = strnlen(
                  reinterpret_cast<const char *>(payload), header.payload_size);
               std::string name(reinterpret_cast<const char *>(payload), length);
               if (pvs_by_name.find(name) != pvs_by_name.end()) {
                  std::vector<std::uint8_t> version;
                  append_u16(version, kProtocolRevision);
                  append_message(reply, kSearch, port, 0, 0xffffffffu,
                                 header.parameter2, version.data(), version.size());
               }
            }
            offset += total;
         }
         if (!reply.empty())
            sendto(udp_fd, reply.data(), reply.size(), 0,
                   reinterpret_cast<sockaddr *>(&source), source_size);
      }
   }

   void accept_clients()
   {
      for (;;) {
         sockaddr_in address;
         socklen_t size = sizeof(address);
         int fd = accept(listen_fd, reinterpret_cast<sockaddr *>(&address), &size);
         if (fd < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
               return;
            return;
         }
         if (!set_nonblocking(fd)) {
            close(fd);
            continue;
         }
         int enabled = 1;
         setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &enabled, sizeof(enabled));
#ifdef SO_NOSIGPIPE
         setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &enabled, sizeof(enabled));
#endif
         std::unique_ptr<Client> client(new (std::nothrow) Client(fd));
         if (!client.get()) {
            close(fd);
            continue;
         }
         append_message(client->output, kVersion, 0, kProtocolRevision, 0, 0);
         clients.push_back(std::move(client));
      }
   }

   static std::uint16_t native_type(const Pv &pv)
   {
      return pv.type == DoubleValue ? DBR_DOUBLE : DBR_LONG;
   }

   static double numeric_value(const Pv &pv)
   {
      if (pv.type == DoubleValue)
         return *static_cast<double *>(pv.value);
      return *static_cast<std::int32_t *>(pv.value);
   }

   static std::int32_t integer_value(const Pv &pv)
   {
      if (pv.type == Int32Value)
         return *static_cast<std::int32_t *>(pv.value);
      return static_cast<std::int32_t>(*static_cast<double *>(pv.value));
   }

   static std::string text_value(const Pv &pv)
   {
      if (pv.type == DoubleValue)
         return value_string(*static_cast<double *>(pv.value));
      return value_string(*static_cast<std::int32_t *>(pv.value));
   }

   static void append_timestamp(std::vector<std::uint8_t> &payload)
   {
      timeval now;
      gettimeofday(&now, NULL);
      append_u16(payload, 0); // alarm status
      append_u16(payload, 0); // alarm severity
      const std::uint32_t seconds = now.tv_sec > kEpicsEpochOffset
         ? static_cast<std::uint32_t>(now.tv_sec - kEpicsEpochOffset) : 0;
      append_u32(payload, seconds);
      append_u32(payload, static_cast<std::uint32_t>(now.tv_usec) * 1000u);
   }

   static int append_plain_value(const Pv &pv, std::uint16_t type,
                                 std::vector<std::uint8_t> &payload)
   {
      switch (type) {
      case DBR_STRING: {
         const std::string text = text_value(pv);
         payload.resize(40, 0);
         std::memcpy(payload.data(), text.data(), std::min<std::size_t>(39, text.size()));
         return ECA_NORMAL;
      }
      case DBR_SHORT:
      case DBR_ENUM:
         append_u16(payload, static_cast<std::uint16_t>(integer_value(pv)));
         return ECA_NORMAL;
      case DBR_FLOAT:
         append_float(payload, static_cast<float>(numeric_value(pv)));
         return ECA_NORMAL;
      case DBR_CHAR:
         payload.push_back(static_cast<std::uint8_t>(integer_value(pv)));
         return ECA_NORMAL;
      case DBR_LONG:
         append_u32(payload, static_cast<std::uint32_t>(integer_value(pv)));
         return ECA_NORMAL;
      case DBR_DOUBLE:
         append_double(payload, numeric_value(pv));
         return ECA_NORMAL;
      default:
         return ECA_BADTYPE;
      }
   }

   static int encode_value(const Pv &pv, std::uint16_t type,
                           std::vector<std::uint8_t> &payload)
   {
      if (type <= DBR_DOUBLE)
         return append_plain_value(pv, type, payload);
      if (type < 14 || type > 20)
         return ECA_BADTYPE;

      const std::uint16_t plain_type = type - 14;
      append_timestamp(payload);
      if (plain_type == DBR_STRING) {
         const std::string text = text_value(pv);
         payload.resize(52, 0);
         std::memcpy(payload.data() + 12, text.data(),
                     std::min<std::size_t>(39, text.size()));
         return ECA_NORMAL;
      }
      if (plain_type == DBR_SHORT || plain_type == DBR_ENUM) {
         append_u16(payload, 0);
         append_u16(payload, static_cast<std::uint16_t>(integer_value(pv)));
      } else if (plain_type == DBR_FLOAT) {
         append_float(payload, static_cast<float>(numeric_value(pv)));
      } else if (plain_type == DBR_CHAR) {
         append_u16(payload, 0);
         payload.push_back(0);
         payload.push_back(static_cast<std::uint8_t>(integer_value(pv)));
      } else if (plain_type == DBR_LONG) {
         append_u32(payload, static_cast<std::uint32_t>(integer_value(pv)));
      } else if (plain_type == DBR_DOUBLE) {
         append_u32(payload, 0);
         append_double(payload, numeric_value(pv));
      }
      return ECA_NORMAL;
   }

   static bool parse_text(const std::uint8_t *payload, std::size_t size,
                          double &value)
   {
      const std::size_t length = strnlen(
         reinterpret_cast<const char *>(payload), size);
      std::string text(reinterpret_cast<const char *>(payload), length);
      char *end = NULL;
      errno = 0;
      value = std::strtod(text.c_str(), &end);
      return end != text.c_str() && *end == 0 && errno != ERANGE &&
             std::isfinite(value);
   }

   static int decode_numeric(std::uint16_t type, const std::uint8_t *payload,
                             std::size_t size, double &value)
   {
      switch (type) {
      case DBR_STRING:
         return parse_text(payload, size, value) ? ECA_NORMAL : ECA_BADSTR;
      case DBR_SHORT:
         if (size < 2) return ECA_BADCOUNT;
         value = static_cast<std::int16_t>(read_u16(payload));
         return ECA_NORMAL;
      case DBR_FLOAT: {
         if (size < 4) return ECA_BADCOUNT;
         std::uint32_t bits = read_u32(payload);
         float number;
         std::memcpy(&number, &bits, sizeof(number));
         value = number;
         return ECA_NORMAL;
      }
      case DBR_ENUM:
         if (size < 2) return ECA_BADCOUNT;
         value = read_u16(payload);
         return ECA_NORMAL;
      case DBR_CHAR:
         if (!size) return ECA_BADCOUNT;
         value = payload[0];
         return ECA_NORMAL;
      case DBR_LONG:
         if (size < 4) return ECA_BADCOUNT;
         value = static_cast<std::int32_t>(read_u32(payload));
         return ECA_NORMAL;
      case DBR_DOUBLE: {
         if (size < 8) return ECA_BADCOUNT;
         std::uint64_t bits = read_u64(payload);
         std::memcpy(&value, &bits, sizeof(value));
         return ECA_NORMAL;
      }
      default:
         return ECA_BADTYPE;
      }
   }

   static int write_value(Pv &pv, std::uint16_t type,
                          const std::uint8_t *payload, std::size_t size)
   {
      if (!pv.writable)
         return ECA_NOWTACCESS;
      double value = 0;
      const int status = decode_numeric(type, payload, size, value);
      if (status != ECA_NORMAL)
         return status;
      if (pv.type == DoubleValue) {
         *static_cast<double *>(pv.value) = value;
      } else {
         if (value < std::numeric_limits<std::int32_t>::min() ||
             value > std::numeric_limits<std::int32_t>::max())
            return ECA_BADSTR;
         *static_cast<std::int32_t *>(pv.value) = static_cast<std::int32_t>(value);
      }
      return ECA_NORMAL;
   }

   static void queue_monitor(Client &client, Monitor &monitor)
   {
      std::vector<std::uint8_t> value;
      const int status = encode_value(*monitor.pv, monitor.data_type, value);
      append_message(client.output, kEventAdd, monitor.data_type,
                     status == ECA_NORMAL ? 1 : 0, status,
                     monitor.subscription_id,
                     value.empty() ? NULL : value.data(), value.size());
      monitor.last_value = numeric_value(*monitor.pv);
   }

   void publish_changes()
   {
      for (std::size_t i = 0; i < clients.size(); ++i) {
         Client &client = *clients[i];
         for (std::unordered_map<std::uint32_t, Monitor>::iterator monitor =
                 client.monitors.begin(); monitor != client.monitors.end(); ++monitor) {
            const double value = numeric_value(*monitor->second.pv);
            if (value != monitor->second.last_value &&
                !(std::isnan(value) && std::isnan(monitor->second.last_value)))
               queue_monitor(client, monitor->second);
         }
      }
   }

   void handle_message(Client &client, const Header &header,
                       const std::uint8_t *payload)
   {
      if (header.command == kCreateChannel) {
         const std::size_t length = strnlen(
            reinterpret_cast<const char *>(payload), header.payload_size);
         const std::string name(reinterpret_cast<const char *>(payload), length);
         std::unordered_map<std::string, Pv *>::iterator found = pvs_by_name.find(name);
         if (found == pvs_by_name.end())
            return;
         Channel channel;
         channel.pv = found->second;
         channel.cid = header.parameter1;
         channel.sid = next_sid++;
         client.channels[channel.sid] = channel;
         append_message(client.output, kAccessRights, 0, 0, channel.cid,
                        channel.pv->writable ? 3 : 1);
         append_message(client.output, kCreateChannel, native_type(*channel.pv), 1,
                        channel.cid, channel.sid);
         return;
      }

      if (header.command == kEcho) {
         append_message(client.output, kEcho, header.data_type, header.data_count,
                        header.parameter1, header.parameter2, payload,
                        header.payload_size);
         return;
      }
      if (header.command == kVersion || header.command == kClientName ||
          header.command == kHostName)
         return;

      std::unordered_map<std::uint32_t, Channel>::iterator found =
         client.channels.find(header.parameter1);
      if (found == client.channels.end())
         return;
      Channel &channel = found->second;

      if (header.command == kReadNotify) {
         std::vector<std::uint8_t> value;
         int status = header.data_count == 1
            ? encode_value(*channel.pv, header.data_type, value) : ECA_BADCOUNT;
         append_message(client.output, kReadNotify, header.data_type,
                        status == ECA_NORMAL ? 1 : 0, status, header.parameter2,
                        value.empty() ? NULL : value.data(), value.size());
      } else if (header.command == kEventAdd) {
         if ((header.data_count == 0 || header.data_count == 1) &&
             (header.data_type <= DBR_DOUBLE ||
              (header.data_type >= 14 && header.data_type <= 20))) {
            Monitor monitor;
            monitor.pv = channel.pv;
            monitor.sid = channel.sid;
            monitor.subscription_id = header.parameter2;
            monitor.data_type = header.data_type;
            monitor.last_value = numeric_value(*channel.pv);
            client.monitors[monitor.subscription_id] = monitor;
            queue_monitor(client, client.monitors[monitor.subscription_id]);
         }
      } else if (header.command == kEventCancel) {
         std::unordered_map<std::uint32_t, Monitor>::iterator monitor =
            client.monitors.find(header.parameter2);
         if (monitor != client.monitors.end() && monitor->second.sid == channel.sid) {
            append_message(client.output, kEventAdd, monitor->second.data_type, 1,
                           channel.sid, monitor->second.subscription_id);
            client.monitors.erase(monitor);
         }
      } else if (header.command == kWrite || header.command == kWriteNotify) {
         int status = header.data_count == 1
            ? write_value(*channel.pv, header.data_type, payload,
                          header.payload_size) : ECA_BADCOUNT;
         if (status == ECA_NORMAL) {
            std::printf("CA write: %s = %s\n", channel.pv->name.c_str(),
                        text_value(*channel.pv).c_str());
            std::fflush(stdout);
         }
         if (header.command == kWriteNotify)
            append_message(client.output, kWriteNotify, header.data_type,
                           header.data_count, status, header.parameter2);
      } else if (header.command == kClearChannel) {
         append_message(client.output, kClearChannel, header.data_type,
                        header.data_count, header.parameter1, header.parameter2);
         for (std::unordered_map<std::uint32_t, Monitor>::iterator monitor =
                 client.monitors.begin(); monitor != client.monitors.end();) {
            if (monitor->second.sid == channel.sid)
               monitor = client.monitors.erase(monitor);
            else
               ++monitor;
         }
         client.channels.erase(found);
      }
   }

   void receive_client(Client &client)
   {
      for (;;) {
         std::uint8_t buffer[8192];
         ssize_t received = recv(client.fd, buffer, sizeof(buffer), 0);
         if (received > 0) {
            client.input.insert(client.input.end(), buffer, buffer + received);
         } else if (received == 0) {
            client.eof = true;
            break;
         } else {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
               break;
            client.dead = true;
            return;
         }
      }

      std::size_t offset = 0;
      while (offset + kHeaderSize <= client.input.size()) {
         Header header;
         if (!decode_header(client.input.data() + offset,
                            client.input.size() - offset, header))
            break;
         if (header.payload_size > kMaxPayloadSize) {
            client.dead = true;
            return;
         }
         const std::size_t total = header.header_size + header.payload_size;
         if (total > client.input.size() - offset)
            break;
         handle_message(client, header,
                        client.input.data() + offset + header.header_size);
         offset += total;
      }
      if (offset)
         client.input.erase(client.input.begin(), client.input.begin() + offset);
   }

   void flush_client(Client &client)
   {
      while (!client.dead && client.output_offset < client.output.size()) {
#ifdef MSG_NOSIGNAL
         const int flags = MSG_NOSIGNAL;
#else
         const int flags = 0;
#endif
         ssize_t sent = send(client.fd, client.output.data() + client.output_offset,
                             client.output.size() - client.output_offset, flags);
         if (sent > 0) {
            client.output_offset += static_cast<std::size_t>(sent);
         } else if (sent < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
            return;
         } else {
            client.dead = true;
            return;
         }
      }
      if (client.output_offset == client.output.size()) {
         client.output.clear();
         client.output_offset = 0;
      }
   }

   bool poll_once(int timeout_ms)
   {
      if (udp_fd < 0 || listen_fd < 0) {
         error = "IOC is not running";
         return false;
      }

      publish_changes();

      std::vector<pollfd> fds;
      std::vector<Client *> active;
      pollfd udp = {udp_fd, POLLIN, 0};
      pollfd listener = {listen_fd, POLLIN, 0};
      fds.push_back(udp);
      fds.push_back(listener);
      for (std::size_t i = 0; i < clients.size(); ++i) {
         Client *client = clients[i].get();
         pollfd item = {client->fd, POLLIN, 0};
         if (client->output_offset < client->output.size())
            item.events |= POLLOUT;
         fds.push_back(item);
         active.push_back(client);
      }

      int result = ::poll(fds.data(), fds.size(), timeout_ms);
      if (result < 0) {
         if (errno == EINTR)
            return true;
         error = socket_error("poll");
         return false;
      }
      if (fds[0].revents & POLLIN)
         receive_discovery();
      if (fds[1].revents & POLLIN)
         accept_clients();
      for (std::size_t i = 0; i < active.size(); ++i) {
         Client &client = *active[i];
         const short events = fds[i + 2].revents;
         if (events & (POLLERR | POLLNVAL))
            client.dead = true;
         if (!client.dead && (events & POLLIN))
            receive_client(client);
         if (events & POLLHUP)
            client.eof = true;
         if (!client.dead && ((events & POLLOUT) ||
                              client.output_offset < client.output.size()))
            flush_client(client);
         if (client.eof && client.output.empty())
            client.dead = true;
      }

      clients.erase(std::remove_if(clients.begin(), clients.end(),
         [](const std::unique_ptr<Client> &client) {
            if (!client->dead)
               return false;
            close(client->fd);
            return true;
         }), clients.end());
      return true;
   }
};

SoftIoc::SoftIoc() : impl_(new Impl)
{
}

SoftIoc::~SoftIoc()
{
}

bool SoftIoc::add_pv(const std::string &name, double *value, bool writable)
{
   return impl_->add(name, value, Impl::DoubleValue, writable);
}

bool SoftIoc::add_pv(const std::string &name, std::int32_t *value, bool writable)
{
   return impl_->add(name, value, Impl::Int32Value, writable);
}

bool SoftIoc::start(std::uint16_t port)
{
   return impl_->start(port);
}

bool SoftIoc::poll(int timeout_ms)
{
   return impl_->poll_once(timeout_ms);
}

const std::string &SoftIoc::error() const
{
   return impl_->error;
}

} // namespace epics
} // namespace midas