git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2023 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.binding.paradoxalarm.internal.communication;
  14
  15 import java.util.concurrent.ArrayBlockingQueue;
  16 import java.util.concurrent.BlockingQueue;
  17
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20
  21 /**
  22  * The {@link SyncQueue} is used to synchronize communication to/from Paradox system. All requests go into sendQueue and
  23  * upon send are popped from send queue and are pushed into receiveQueue.
  24  * Due to nature of Paradox communication receive queue is with priority, i.e. if there is anything in receive queue we
  25  * attempt to read the socket first and only after receive queue is empty then we attempt to send. We never send any
  26  * packet if we have something to read.
  27  * For more details about usage see method {@link AbstractCommunicator.submitRequest()}
  28  *
  29  * @author Konstantin Polihronov - Initial contribution
  30  */
  31 public class SyncQueue {
  32
  33     private final Logger logger = LoggerFactory.getLogger(SyncQueue.class);
  34
  35     private BlockingQueue<IRequest> sendQueue = new ArrayBlockingQueue<>(1000, true);
  36     private BlockingQueue<IRequest> receiveQueue = new ArrayBlockingQueue<>(10, true);
  37
  38     private static SyncQueue syncQueue;
  39
  40     private SyncQueue() {
  41     }
  42
  43     public static SyncQueue getInstance() {
  44         SyncQueue temp = syncQueue;
  45         if (temp == null) {
  46             synchronized (SyncQueue.class) {
  47                 temp = syncQueue;
  48                 if (temp == null) {
  49                     syncQueue = new SyncQueue();
  50                 }
  51             }
  52         }
  53         return syncQueue;
  54     }
  55
  56     public synchronized void add(IRequest request) {
  57         logger.trace("Adding to queue request={}", request);
  58         sendQueue.add(request);
  59     }
  60
  61     public synchronized void moveRequest() {
  62         IRequest request = sendQueue.poll();
  63         request.setTimeStamp();
  64         logger.trace("Moving from Tx to RX queue request={}", request);
  65         receiveQueue.add(request);
  66     }
  67
  68     public synchronized IRequest poll() {
  69         IRequest request = receiveQueue.poll();
  70         logger.trace("Removing from queue request={}", request);
  71         return request;
  72     }
  73
  74     public synchronized IRequest removeSendRequest() {
  75         IRequest request = sendQueue.poll();
  76         logger.trace("Removing from queue request={}", request);
  77         return request;
  78     }
  79
  80     public synchronized IRequest peekSendQueue() {
  81         return sendQueue.peek();
  82     }
  83
  84     public IRequest peekReceiveQueue() {
  85         return receiveQueue.peek();
  86     }
  87
  88     public synchronized boolean hasPacketToReceive() {
  89         return receiveQueue.peek() != null;
  90     }
  91
  92     public synchronized boolean hasPacketsToSend() {
  93         return sendQueue.peek() != null;
  94     }
  95
  96     public synchronized boolean canSend() {
  97         return receiveQueue.isEmpty();
  98     }
  99 }