IncomingMessageQueue.java

Go to the documentation of this file.

/*
 * Copyright (C) 2011 Google Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package org.ros.internal.transport.queue;

import org.ros.concurrent.CircularBlockingDeque;
import org.ros.internal.transport.tcp.NamedChannelHandler;
import org.ros.message.MessageDeserializer;
import org.ros.message.MessageListener;

import java.util.concurrent.ExecutorService;

public class IncomingMessageQueue<T> {

  private static final int DEQUE_CAPACITY = 16;

  private final MessageReceiver<T> messageReceiver;
  private final MessageDispatcher<T> messageDispatcher;

  public IncomingMessageQueue(MessageDeserializer<T> deserializer, ExecutorService executorService) {
    CircularBlockingDeque<LazyMessage<T>> lazyMessages =
        new CircularBlockingDeque<LazyMessage<T>>(DEQUE_CAPACITY);
    messageReceiver = new MessageReceiver<T>(lazyMessages, deserializer);
    messageDispatcher = new MessageDispatcher<T>(lazyMessages, executorService);
    executorService.execute(messageDispatcher);
  }

  public void setLatchMode(boolean enabled) {
    messageDispatcher.setLatchMode(enabled);
  }

  public boolean getLatchMode() {
    return messageDispatcher.getLatchMode();
  }

  public void addListener(final MessageListener<T> messageListener, int queueCapacity) {
    messageDispatcher.addListener(messageListener, queueCapacity);
  }

  public void shutdown() {
    messageDispatcher.cancel();
  }

  public NamedChannelHandler getMessageReceiver() {
    return messageReceiver;
  }
}