KLALB/src/org/kne/cloud/network/IPMulticastDiscovery.java

package org.kne.cloud.network;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.*;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.function.Consumer;

public class IPMulticastDiscovery implements Closeable, AutoCloseable {
	private static final boolean debug=false;
	
	private MulticastSocket soc;
	private NetworkInterface ninterface;
	private volatile boolean closed = false;
	private InetSocketAddress group;
	private int type;
	private List<MultiProtocolSocketAddress> msas;
	private UUID selfUUID;

	private volatile Consumer<MultiProtocolSocketAddress> con;
	private long timeInterval;
	
	public long getTimeInterval() {
		return timeInterval;
	}

	public void setTimeInterval(long timeInterval) {
		this.timeInterval = timeInterval;
	}

	public NetworkInterface getInterface() {
		return ninterface;
	}

	public InetSocketAddress getGroup() {
		return group;
	}

	public InetSocketAddress getBind() {
		return (InetSocketAddress) soc.getLocalSocketAddress();
	}

	public Consumer<MultiProtocolSocketAddress> getCon() {
		return con;
	}

	public void setCon(Consumer<MultiProtocolSocketAddress> con) {
		this.con = con;
	}

	public IPMulticastDiscovery(InetSocketAddress bind, InetSocketAddress group, NetworkInterface ninterface,
                                List<MultiProtocolSocketAddress> msas, UUID node, long timeInterval) throws IOException {
		soc = new MulticastSocket(null);
		soc.setReuseAddress(true);
		soc.bind(bind);
		soc.setNetworkInterface(ninterface);
		soc.joinGroup(group, ninterface);
		this.group = group;
		this.ninterface = ninterface;
		this.msas = msas;
		this.selfUUID=node;
		this.timeInterval=timeInterval;
	}

	public void start() {

		ThreadTool.makeVDaemonThreadIfSupport("线路通知线程", () -> {
			try {
				while (!closed) {
						/*MultipurposeSocketAddress mpsa = new MultipurposeSocketAddress(
								multipurposeSocketAddress.getType(), "::0", multipurposeSocketAddress.getPort());*/
					if(msas!=null) {
						synchronized (msas) {
							for (Iterator<MultiProtocolSocketAddress> iterator = msas.iterator(); iterator.hasNext(); ) {
								MultiProtocolSocketAddress multiProtocolSocketAddress = (MultiProtocolSocketAddress) iterator
										.next();

								try {
										send(multiProtocolSocketAddress);
								} catch (NoRouteToHostException | UnknownHostException e) {
								}

							}
						}
					}
					Thread.sleep(timeInterval);
				}
			} catch (IOException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO 自动生成的 catch 块
				e.printStackTrace();
			} finally {
				try {
					close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}

		}).start();

		ThreadTool.makeVDaemonThreadIfSupport("线路发现线程",()->{
		try {
			loop:while (!closed) {
				try {
					MultiProtocolSocketAddress mpsa = receive();
					if (mpsa == null) continue;
					InetAddress mpsai=mpsa.getInetAddress();
				Enumeration<InetAddress> ei = ninterface.getInetAddresses();
				while (ei.hasMoreElements()) {
					InetAddress inetAddress = (InetAddress) ei.nextElement();
					if (inetAddress.equals(mpsai)) {
						continue loop;
					}
				}
				
				if (con != null) {
					con.accept(mpsa);
				}
				}catch(UnknownHostException e) {
					
				}catch(EOFException e) {
					
				}
			}
		} catch(SocketException e) {
			if(!isClosed())
				e.printStackTrace();
		}catch (IOException e) {
			System.out.println(getBind() + " " + group + " " + ninterface);
			e.printStackTrace();
		} finally {
			try {
				
				close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		}).start();
	}

	private MultiProtocolSocketAddress receive() throws IOException {
		byte[] b = new byte[65535];
		DatagramPacket dp = new DatagramPacket(b, b.length);
		soc.receive(dp);
		ByteArrayInputStream bis = new ByteArrayInputStream(b, 0, dp.getLength());
		DataInputStream dis = new DataInputStream(bis);
		String val = dis.readUTF();
		long h=dis.readLong();
		long l=dis.readLong();
		UUID uid=new UUID(h, l);
		dis.close();
		MultiProtocolSocketAddress mpsa = new MultiProtocolSocketAddress(val);
		if(mpsa.getInetAddress().isAnyLocalAddress()){
			mpsa=new MultiProtocolSocketAddress(mpsa.getProtocol(),dp.getAddress().getHostAddress(),mpsa.getPort());
		}
		if(debug) {
			if(uid.equals(selfUUID)) {
				System.out.println(ninterface.getDisplayName()+" 丢弃广播："+mpsa+" "+dp.getSocketAddress()+" "+uid);
				return null;
			}else {
				System.out.println(ninterface.getDisplayName()+ " 接收广播："+mpsa+" "+dp.getSocketAddress()+" "+uid);
			}
		}
		return mpsa;
	}

	private void send(MultiProtocolSocketAddress multiProtocolSocketAddress) throws IOException {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		DataOutputStream dops = new DataOutputStream(baos);
		dops.writeUTF(multiProtocolSocketAddress.toString());
		dops.writeLong(selfUUID.getMostSignificantBits());
		dops.writeLong(selfUUID.getLeastSignificantBits());
		dops.close();
		byte[] b = baos.toByteArray();
		DatagramPacket dp = new DatagramPacket(b, b.length, group);
		soc.send(dp);
		if(debug)
			System.out.println(ninterface.getDisplayName()+" 发送广播："+ multiProtocolSocketAddress +" "+group+" "+selfUUID);

	}

	@Override
	public void close() throws IOException {
		soc.close();
		closed = true;
	}

	public boolean isClosed() {
		return closed;
	}

}