package org.jgroups.protocols;

import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.Event;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.View;
import org.junit.Test;

/**
 * Demonstration of the problem with NAKACK's window creation.
 * 
 * @author graywatson
 */
public class NAKACKEarlyTest {

	@Test
	public void testQuickSend() throws Throwable {
		Channel ch0 = null;
		Channel ch1 = null;
		try {
			ch0 = new JChannel("udp.xml");
			ch1 = new JChannel("udp.xml");
			OurReceiver rec0 = startReceiver(ch0);
			OurReceiver rec1 = startReceiver(ch1);
			assertTrue(rec0.worked);
			assertTrue(rec1.worked);
		} finally {
			if (ch0 != null)
				ch0.close();
			if (ch1 != null)
				ch1.close();
		}
	}

	private OurReceiver startReceiver(Channel channel) throws Throwable {
		OurReceiver receiver = new OurReceiver(channel);
		channel.setReceiver(receiver);
		channel.connect("foo");
		return receiver;
	}

	private class OurReceiver implements Receiver {
		private Channel channel;
		private boolean worked = true;
		public OurReceiver(Channel channel) {
			this.channel = channel;
		}
		public void receive(Message msg) {
			if (msg.getDest() == null) {
				System.out.println(channel.getLocalAddress() + ": received multicast message from remote "
						+ msg.getSrc());
			} else {
				System.out.println(channel.getLocalAddress() + ": received unicast message from remote " + msg.getSrc());
				// send out a broadcast response
				try {
					channel.down(new Event(Event.MSG, new Message(null)));
				} catch (Exception e) {
					e.printStackTrace();
					worked = false;
				}
			}
		}
		public byte[] getState() {
			return null;
		}
		public void setState(byte[] state) {
		}
		public void block() {
		}
		public void suspect(Address suspected_mbr) {
		}
		public void viewAccepted(View new_view) {
			// as soon as we get new members, investigate their status
			for (Address address : new_view.getMembers()) {
				channel.down(new Event(Event.MSG, new Message(address)));
			}
		}
	}
}