/*******************************************************************************
 * Copyright 2018 Klaus Pfeiffer - klaus@allpiper.com
 *
 * 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 com.jfastnet;

import com.jfastnet.messages.Message;
import com.jfastnet.processors.MessageLogProcessor;
import com.jfastnet.util.FifoMap;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.queue.CircularFifoQueue;

import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;

/** Logs incoming and outgoing messages. Per default only reliable messages
 * get logged and an upper bound for the message log is used.
 *
 * The reliable sequence processor will grab required messages from the message
 * log. So don't change the filter unless you know what you're doing.
 * @author Klaus Pfeiffer - klaus@allpiper.com */
@Slf4j
public class MessageLog {

	@Getter
	private CircularFifoQueue<Message> received;

	/** First-in, first-out map of sent messages. If a missing id is requested
	 * by another peer, this map gets queried for it. */
	private FifoMap<MessageKey, Message> sentMap;

	private final ReentrantLock sentMapLock = new ReentrantLock();

	private Config config;

	private MessageLogProcessor.ProcessorConfig processorConfig;

	public MessageLog(Config config, MessageLogProcessor.ProcessorConfig processorConfig) {
		this.config = config;
		this.processorConfig = processorConfig;
		received = new CircularFifoQueue<>(processorConfig.receivedMessagesLimit);
		sentMap = new FifoMap<>(processorConfig.sentMessagesMapLimit);
	}

	public synchronized void addReceived(Message message) {
		if (processorConfig.messageLogReceiveFilter.test(message)) {
			received.add(message);
		}
	}

	public Message getSent(MessageKey key) {
		sentMapLock.lock();
		try {
			return sentMap.get(key);
		} finally {
			sentMapLock.unlock();
		}
	}

	public void addSent(Message message) {
		if (!message.isResendMessage() && processorConfig.messageLogSendFilter.test(message)) {
			MessageKey messageKey = MessageKey.newKey(message.getReliableMode(), message.getReceiverId(), message.getMsgId());
			if (sentMap.containsKey(messageKey)) {
				//log.trace("Message already in map! Skipping!");
				return;
			}
			sentMapLock.lock();
			try {
				log.trace("Put into sent-log: {} -- {}", messageKey, message);
				sentMap.put(messageKey, message);
			} finally {
				sentMapLock.unlock();
			}
		}
	}

	/** All messages are logged. */
	public static class AllMessagesPredicate implements Predicate<Message> {
		@Override
		public boolean test(Message message) {
			return true;
		}
	}

	/** No messages are logged. */
	public static class NoMessagesPredicate implements Predicate<Message> {
		@Override
		public boolean test(Message message) {
			return false;
		}
	}

	/** Only reliable messages are logged. */
	public static class ReliableMessagesPredicate implements Predicate<Message> {
		@Override
		public boolean test(Message message) {
			return !Message.ReliableMode.UNRELIABLE.equals(message.getReliableMode());
		}
	}

}

@@ -1070,25 +1070,123 @@
 t.locks.
-Reentrant
+Lock;%0Aimport java.util.concurrent.locks.ReadWriteLock;%0Aimport java.util.concurrent.locks.ReentrantReadWrite
 Lock;%0Aim
@@ -1847,23 +1847,23 @@
 final Re
-entrant
+adWrite
 Lock sen
@@ -1886,16 +1886,25 @@
 eentrant
+ReadWrite
 Lock();%0A
@@ -2491,23 +2491,62 @@
 ey) %7B%0A%09%09
-sentMap
+Lock readLock = sentMapLock.readLock();%0A%09%09read
 Lock.loc
@@ -2599,31 +2599,28 @@
 inally %7B%0A%09%09%09
-sentMap
+read
 Lock.unlock(
@@ -2878,16 +2878,89 @@
 gId());%0A
+%09%09%09Lock readLock = sentMapLock.readLock();%0A%09%09%09readLock.lock();%0A%09%09%09try %7B%0A%09
 %09%09%09if (s
@@ -2993,16 +2993,17 @@
 Key)) %7B%0A
+%09
 %09%09%09%09//lo
@@ -3056,16 +3056,17 @@
 %09%09%09%09
+%09
 return;%0A
 %09%09%09%7D
@@ -3065,16 +3065,17 @@
 rn;%0A
+%09
 %09%09%09%7D%0A%09%09%09
 sent
@@ -3070,23 +3070,109 @@
 %09%09%09%7D%0A%09%09%09
-sentMap
+%7D finally %7B%0A%09%09%09%09readLock.unlock();%0A%09%09%09%7D%0A%09%09%09Lock writeLock = sentMapLock.writeLock();%0A%09%09%09write
 Lock.loc
@@ -3309,23 +3309,21 @@
 y %7B%0A%09%09%09%09
-sentMap
+write
 Lock.unl

use of ReadWriteLock for MessageLog