@Override
public MembershipKey block(InetAddress toBlock)
    throws IOException
{
    if (source != null)
        throw new IllegalStateException("key is source-specific");

    synchronized (stateLock) {
        if ((blockedSet != null) && blockedSet.contains(toBlock)) {
            // already blocked, nothing to do
            return this;
        }

        ((DatagramChannelImpl)ch).block(this, toBlock);

        // created blocked set if required and add source address
        if (blockedSet == null)
            blockedSet = new HashSet<>();
        blockedSet.add(toBlock);
    }
    return this;
}
 

@Override
public ChannelFuture leaveGroup(
        InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
        ChannelPromise promise) {
    checkJavaVersion();

    if (multicastAddress == null) {
        throw new NullPointerException("multicastAddress");
    }
    if (networkInterface == null) {
        throw new NullPointerException("networkInterface");
    }

    synchronized (this) {
        if (memberships != null) {
            List<MembershipKey> keys = memberships.get(multicastAddress);
            if (keys != null) {
                Iterator<MembershipKey> keyIt = keys.iterator();

                while (keyIt.hasNext()) {
                    MembershipKey key = keyIt.next();
                    if (networkInterface.equals(key.networkInterface())) {
                       if (source == null && key.sourceAddress() == null ||
                           source != null && source.equals(key.sourceAddress())) {
                           key.drop();
                           keyIt.remove();
                       }
                    }
                }
                if (keys.isEmpty()) {
                    memberships.remove(multicastAddress);
                }
            }
        }
    }

    promise.setSuccess();
    return promise;
}
 

/**
 * Block the given sourceToBlock address for the given multicastAddress on the given networkInterface
 */
@Override
public ChannelFuture block(
        InetAddress multicastAddress, NetworkInterface networkInterface,
        InetAddress sourceToBlock, ChannelPromise promise) {
    checkJavaVersion();

    if (multicastAddress == null) {
        throw new NullPointerException("multicastAddress");
    }
    if (sourceToBlock == null) {
        throw new NullPointerException("sourceToBlock");
    }

    if (networkInterface == null) {
        throw new NullPointerException("networkInterface");
    }
    synchronized (this) {
        if (memberships != null) {
            List<MembershipKey> keys = memberships.get(multicastAddress);
            for (MembershipKey key: keys) {
                if (networkInterface.equals(key.networkInterface())) {
                    try {
                        key.block(sourceToBlock);
                    } catch (IOException e) {
                        promise.setFailure(e);
                    }
                }
            }
        }
    }
    promise.setSuccess();
    return promise;
}
 

/**
 * Checks registry for membership of the group on the given
 * network interface.
 */
MembershipKey checkMembership(InetAddress group, NetworkInterface interf,
                              InetAddress source)
{
    if (groups != null) {
        List<MembershipKeyImpl> keys = groups.get(group);
        if (keys != null) {
            for (MembershipKeyImpl key: keys) {
                if (key.networkInterface().equals(interf)) {
                    // already a member to receive all packets so return
                    // existing key or detect conflict
                    if (source == null) {
                        if (key.sourceAddress() == null)
                            return key;
                        throw new IllegalStateException("Already a member to receive all packets");
                    }

                    // already have source-specific membership so return key
                    // or detect conflict
                    if (key.sourceAddress() == null)
                        throw new IllegalStateException("Already have source-specific membership");
                    if (source.equals(key.sourceAddress()))
                        return key;
                }
            }
        }
    }
    return null;
}
 

@Override
public MembershipKey join(InetAddress group,
                          NetworkInterface interf)
    throws IOException
{
    return innerJoin(group, interf, null);
}
 

@Override
public MembershipKey join(InetAddress group,
                          NetworkInterface interf,
                          InetAddress source)
    throws IOException
{
    Objects.requireNonNull(source);
    return innerJoin(group, interf, source);
}
 

@Override
public MembershipKey unblock(InetAddress toUnblock) {
    synchronized (stateLock) {
        if ((blockedSet == null) || !blockedSet.contains(toUnblock))
            throw new IllegalStateException("not blocked");

        ((DatagramChannelImpl)ch).unblock(this, toUnblock);

        blockedSet.remove(toUnblock);
    }
    return this;
}
 

@Override
public ChannelFuture leaveGroup(
        InetAddress multicastAddress, NetworkInterface networkInterface, InetAddress source,
        ChannelPromise promise) {
    checkJavaVersion();

    if (multicastAddress == null) {
        throw new NullPointerException("multicastAddress");
    }
    if (networkInterface == null) {
        throw new NullPointerException("networkInterface");
    }

    synchronized (this) {
        if (memberships != null) {
            List<MembershipKey> keys = memberships.get(multicastAddress);
            if (keys != null) {
                Iterator<MembershipKey> keyIt = keys.iterator();

                while (keyIt.hasNext()) {
                    MembershipKey key = keyIt.next();
                    if (networkInterface.equals(key.networkInterface())) {
                       if (source == null && key.sourceAddress() == null ||
                           source != null && source.equals(key.sourceAddress())) {
                           key.drop();
                           keyIt.remove();
                       }
                    }
                }
                if (keys.isEmpty()) {
                    memberships.remove(multicastAddress);
                }
            }
        }
    }

    promise.setSuccess();
    return promise;
}
 

/**
 * Block the given sourceToBlock address for the given multicastAddress on the given networkInterface
 */
@Override
public ChannelFuture block(
        InetAddress multicastAddress, NetworkInterface networkInterface,
        InetAddress sourceToBlock, ChannelPromise promise) {
    checkJavaVersion();

    if (multicastAddress == null) {
        throw new NullPointerException("multicastAddress");
    }
    if (sourceToBlock == null) {
        throw new NullPointerException("sourceToBlock");
    }

    if (networkInterface == null) {
        throw new NullPointerException("networkInterface");
    }
    synchronized (this) {
        if (memberships != null) {
            List<MembershipKey> keys = memberships.get(multicastAddress);
            for (MembershipKey key: keys) {
                if (networkInterface.equals(key.networkInterface())) {
                    try {
                        key.block(sourceToBlock);
                    } catch (IOException e) {
                        promise.setFailure(e);
                    }
                }
            }
        }
    }
    promise.setSuccess();
    return promise;
}
 

@Override
public ChannelFuture joinGroup(
        InetAddress multicastAddress, NetworkInterface networkInterface,
        InetAddress source, ChannelPromise promise) {

    checkJavaVersion();

    if (multicastAddress == null) {
        throw new NullPointerException("multicastAddress");
    }

    if (networkInterface == null) {
        throw new NullPointerException("networkInterface");
    }

    try {
        MembershipKey key;
        if (source == null) {
            key = javaChannel().join(multicastAddress, networkInterface);
        } else {
            key = javaChannel().join(multicastAddress, networkInterface, source);
        }

        synchronized (this) {
            List<MembershipKey> keys = null;
            if (memberships == null) {
                memberships = new HashMap<InetAddress, List<MembershipKey>>();
            } else {
                keys = memberships.get(multicastAddress);
            }
            if (keys == null) {
                keys = new ArrayList<MembershipKey>();
                memberships.put(multicastAddress, keys);
            }
            keys.add(key);
        }

        promise.setSuccess();
    } catch (Throwable e) {
        promise.setFailure(e);
    }

    return promise;
}
 

@Override
public ChannelFuture joinGroup(
        InetAddress multicastAddress, NetworkInterface networkInterface,
        InetAddress source, ChannelPromise promise) {

    checkJavaVersion();

    if (multicastAddress == null) {
        throw new NullPointerException("multicastAddress");
    }

    if (networkInterface == null) {
        throw new NullPointerException("networkInterface");
    }

    try {
        MembershipKey key;
        if (source == null) {
            key = javaChannel().join(multicastAddress, networkInterface);
        } else {
            key = javaChannel().join(multicastAddress, networkInterface, source);
        }

        synchronized (this) {
            List<MembershipKey> keys = null;
            if (memberships == null) {
                memberships = new HashMap<InetAddress, List<MembershipKey>>();
            } else {
                keys = memberships.get(multicastAddress);
            }
            if (keys == null) {
                keys = new ArrayList<MembershipKey>();
                memberships.put(multicastAddress, keys);
            }
            keys.add(key);
        }

        promise.setSuccess();
    } catch (Throwable e) {
        promise.setFailure(e);
    }

    return promise;
}
 

@Override
public MembershipKey join(InetAddress inetAddress, NetworkInterface networkInterface) throws IOException {
    return channel.join(inetAddress, networkInterface);
}
 

@Override
public MembershipKey join(InetAddress inetAddress, NetworkInterface networkInterface, InetAddress inetAddress1) throws IOException {
    return channel.join(inetAddress, networkInterface, inetAddress1);
}
 

@Override
public MembershipKey join(InetAddress inetAddress, NetworkInterface networkInterface) throws IOException {
    throw new UnsupportedOperationException("Not supported yet.");
}
 

@Override
public MembershipKey join(InetAddress inetAddress, NetworkInterface networkInterface, InetAddress inetAddress1) throws IOException {
    throw new UnsupportedOperationException("Not supported yet.");
}