public function postAction()
 {
     $line = $_REQUEST['msg'];
     $id = $_REQUEST['id'];
     $msg_id = msg_get_queue($this->_key, 0600);
     die(json_encode(msg_send($msg_id, 1, "{$id}!{$line}", true, true, $msg_err)));
 }

 /**
  * @param $strMessage   - message to send to the queue
  * @param $intErrorCode - reference to error-code, which is set in case of error
  *
  * @throw \Exception - if message is not a string
  *
  * @returns (boolean) true, if message was successfully send to queue
  * @returns (boolean) false, if error occured
  *
  * send the given message to the queue. if the message is to big,
  * the sending is blocked, until there is enough space in queue
  * to send the message.
  * 
  * if an error occurs, the parameter $intErrorCode will contain
  * the error-code.
  *
  **/
 public function send($strMessage, &$intErrorCode = null)
 {
     if (!is_string($strMessage)) {
         throw new \Exception("given parameter must be a string");
     }
     return msg_send($this->resQueue, 1, $strMessage, false, true, $intErrorCode);
 }

 public function send($message, $type = 1)
 {
     if (!@msg_send($this->queue_id, $type, $message, true, false, $err)) {
         throw new Exception("From msg_send: {$err}");
     }
     return true;
 }

 public function shutdown()
 {
     $this->queue = msg_get_queue(ftok($this->file, 'R'), 0777);
     for ($i = 0; $i < $this->numThreads * 2; $i++) {
         msg_send($this->queue, 100 + $i, 'shutdown');
     }
 }

 public static function log($message, $type = '')
 {
     $contents = trim($message);
     if (!$contents) {
         return false;
     }
     if (empty($type)) {
         $type = 'common';
     }
     $queue = self::getInstance();
     $msg = json_encode(array('msg' => $contents, 'type' => $type));
     msg_send($queue, 1, $msg, false, false, $errorcode);
     $unix_time = time();
     $Y = date("Y", $unix_time);
     $m = date("m", $unix_time);
     $d = date("d", $unix_time);
     $H = date("H", $unix_time);
     $log_dir = self::LC_LOG_PATH . "/" . $type . "/{$Y}/{$m}/{$d}";
     if (!is_dir($log_dir)) {
         @mkdir($log_dir, 0777, true);
     }
     $log = $log_dir . "/" . $type . '.' . $H . '.' . "log";
     file_put_contents($log, "{$contents}\n", FILE_APPEND);
     return true;
 }

 /**
  * {@inheritdoc}
  */
 public function push($item, $eta = null)
 {
     $eta = QueueUtils::normalizeEta($eta);
     if (!msg_send($this->getQueue(), $eta, $item, $this->serialize, false, $errorCode)) {
         throw new QueueException($this, self::getErrorMessage($errorCode), $errorCode);
     }
 }

 function put($data)
 {
     $this->init();
     if (!@msg_send($this->seg, 1, serialize($data), false, false, $errno)) {
         throw new Scalr_System_Ipc_Exception($errno ? self::$msgsnd_errors[$errno] : "Cannot send message", $errno);
     }
     return true;
 }

 public function send($message)
 {
     $result = msg_send($this->queue, 1, $message);
     if ($result == False) {
         print "end error" . PHP_EOL;
     }
     return $result;
 }

 /**
     Put a Message Into the Queue
 */
 static function mqSend($name, $type, $mess)
 {
     $mq = msg_get_queue(self::_ftok($name));
     $block = true;
     $error = null;
     if (msg_send($mq, $type, $mess, true, $block, $error)) {
         return true;
     }
 }

 public function send($data, $block = false, $serialize = false)
 {
     $error = 0;
     $type = 1;
     $r = msg_send($this->msg, $type, $data, $serialize, $block, $error);
     if (!$r) {
     }
     return $r;
 }

 /**
  * Sends message
  * @param  integer $type
  * @param  string  $msg
  * @return void
  */
 public function send($type, $msg)
 {
     $err = 0;
     if (!msg_send($this->queue, $type, $msg, true, false, $err)) {
         $this->logger->critical("Message of type {$type} not sent: {$err}", (array) $msg);
     } else {
         $this->logger->debug("Message of type {$type} sent", (array) $msg);
     }
 }

 public function putMessage(&$message)
 {
     $messageId = $this->storage->saveMessage($message);
     // notify queue via ipc
     $ipcStat = msg_stat_queue($this->ipcChannel);
     if (is_array($ipcStat) && $ipcStat['msg_qnum'] < 5) {
         msg_send($this->ipcChannel, 1, '*', false, false, $ipcError);
     }
     return $messageId;
 }

 /**
  * addMessage: Given a key, store a new message into our queue.
  *
  * @param $key string - Reference to the message (PK)
  * @param $data array - Some data to pass into the message
  */
 public static function addMessage($key, $data = array())
 {
     # What to send
     $message = new Message($key, $data);
     # Try to send the message
     if (msg_send(self::$queue, QUEUE_TYPE_START, $message)) {
         print_r(msg_stat_queue(self::$queue));
     } else {
         echo "Error adding to the queue";
     }
 }

 public static function send($queue, $msg = '')
 {
     $queue_id = self::get_queue_id($queue);
     if (!$queue_id) {
         return false;
     }
     $mqueue = self::get_msg_queue($queue_id);
     if (!$mqueue) {
         return false;
     }
     return msg_send($mqueue, 1, $msg, false, false);
 }

 public function sendMessage($data)
 {
     //Generate a random ID for this request
     $id = rand();
     $message = [$id, $data];
     msg_send($this->queue, $this->serverId, $message, true, false);
     msg_receive($this->queue, $id, $msgtype, 1000000, $msg, true);
     foreach ($msg[0] as $header) {
         header($header);
     }
     return $msg[1];
 }

 public function handle()
 {
     $messageQueueKey = ftok(App::path('cache') . "/queue/daemon.queue", "a");
     $messageQueue = msg_get_queue($messageQueueKey, 0666);
     for ($i = 0; $i < 60; $i++) {
         $message = new MessageQueue();
         $message->pid = $this->getPID();
         $message->name = "IdleDaemon";
         $message->timestamp = time();
         msg_send($messageQueue, 1, $message);
         sleep(1);
     }
 }

 function printt()
 {
     sem_acquire($this->msgSemKey);
     $mode = $this->settings['error_mode']['mode'];
     if ($this->id == 0) {
         if ($mode == 'ERROR') {
             sem_release($this->msgSemKey);
             return;
         } else {
             if ($mode == 'WARNING') {
                 sem_release($this->msgSemKey);
                 return;
             } else {
                 if ($mode == 'INFO') {
                     sem_release($this->msgSemKey);
                     return;
                 }
             }
         }
     } else {
         if ($this->id == 1) {
             if ($mode == 'ERROR') {
                 sem_release($this->msgSemKey);
                 return;
             } else {
                 if ($mode == 'WARNING') {
                     sem_release($this->msgSemKey);
                     return;
                 }
             }
         } else {
             if ($this->id == 3) {
                 if ($mode == 'ERROR') {
                     sem_release($this->msgSemKey);
                     return;
                 }
             }
         }
     }
     $x = $this->info;
     $time = microtime(true);
     $dFormat = "m/d/Y - H:i:s:";
     $mSecs = $time - floor($time);
     $mSecs = substr($mSecs, 2, 4);
     $date = sprintf('%s%s', date($dFormat), $mSecs);
     $type = $this->settings['error_types'][$this->id];
     msg_send($this->queKey, 1, "{$date} {$type} {$x}\n");
     sem_release($this->msgSemKey);
     $this->checkConsumer();
 }

 /**
  * Sends a message to a message queue.
  *
  * @param Message $message   A Message instance
  * @param bool    $serialize If true, $message will be serialized
  * @param bool    $blocking  If true, the process will be blocked until another process reads messages
  *                           from the queue and frees enough space for your message to be sent,
  *                           otherwise will be thrown OverflowException
  *
  * @throws \OverflowException When a message queue is overwoled
  * @throws \RuntimeException  When could not send a $message
  */
 public function send(Message $message, $serialize = true, $blocking = true)
 {
     $this->checkState();
     if (!@msg_send($this->handle, $message->getType(), $message->getData(), $serialize, $blocking, $errCode)) {
         if ($errCode === MSG_EAGAIN) {
             throw new \OverflowException(sprintf('Overflow a message queue with key "%s", not enough the space for a new message.', $this->key));
         }
         $detail = null;
         if ($errCode === 22) {
             $detail = 'Maybe the message size is too big.';
         }
         throw new \RuntimeException(sprintf('Could not send the message to a message queue with key "%s" (error code: %d). %s', $this->key, $errCode, $detail));
     }
 }

 public function run()
 {
     $message_queue_key = ftok(__FILE__, 'a');
     $message_queue = msg_get_queue($message_queue_key, 0666);
     do {
         if (($msgsock = socket_accept($this->socket)) < 0) {
             $this->addLog("socket_accept() failed! reason: " . socket_strerror($msgsock));
             break;
         } else {
             $zhanArr = array();
             $this->addLog("parent start, pid:" . getmypid());
             for ($i = 0; $i < 3; $i++) {
                 $pid = pcntl_fork();
                 if ($pid == -1) {
                     die("cannot fork");
                 } else {
                     if ($pid > 0) {
                         $this->addLog("parent continue, pid:" . getmypid());
                         while (true) {
                             $buf = socket_read($msgsock, 10000);
                             if (!empty($buf)) {
                                 $sendRet = msg_send($message_queue, 1, $buf);
                                 if ($sendRet) {
                                     $this->addLog("send msg to queue ok!send message : " . $buf);
                                 } else {
                                     $this->addLog("send msg to queue failed!!!");
                                 }
                             } else {
                                 $this->addLog("no message");
                                 sleep(2);
                             }
                         }
                     } else {
                         if ($pid == 0) {
                             while (true) {
                                 msg_receive($message_queue, 0, $message_type, 1024, $message, true, MSG_IPC_NOWAIT);
                                 if (!empty($message)) {
                                     $this->addLog("receive message:" . $message);
                                     $this->writeFile(WORK_PATH . '/abc.txt', $message);
                                 } else {
                                     sleep(10);
                                 }
                             }
                         }
                     }
                 }
             }
         }
     } while (true);
 }

 public function publish($message)
 {
     if (empty($message)) {
         throw new Exception\EmptyMessageException();
     }
     $queueResource = $this->queue->getResource();
     $serializeMessage = true;
     $isBlocking = true;
     $errorCode = 0;
     $result = @msg_send($queueResource, self::PRODUCER_DEFAULT_MESSAGE_TYPE, $message, $serializeMessage, $isBlocking, $errorCode);
     if (false === $result) {
         $this->throwExceptionForErrorCode($errorCode);
     }
     return true;
 }