X-Git-Url: https://git.pterodactylus.net/?p=xudocci.git;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Fxdcc%2Fcore%2FCore.java;h=d39ee2bd0ffe241a933ad8ae0c844f99c21a93f4;hp=eb0036c92d9c904fe45ea5aaf3300617b42b7faf;hb=d8c6d72caaa1ba75d09bd28f2bb10ae6f4964bf4;hpb=f724f1d4c327876f6a296a976b58e413c1eaaed6 diff --git a/src/main/java/net/pterodactylus/xdcc/core/Core.java b/src/main/java/net/pterodactylus/xdcc/core/Core.java index eb0036c..d39ee2b 100644 --- a/src/main/java/net/pterodactylus/xdcc/core/Core.java +++ b/src/main/java/net/pterodactylus/xdcc/core/Core.java @@ -17,6 +17,10 @@ package net.pterodactylus.xdcc.core; +import static java.lang.String.format; +import static net.pterodactylus.irc.event.ChannelNotJoined.Reason.banned; +import static net.pterodactylus.irc.event.ChannelNotJoined.Reason.inviteOnly; +import static net.pterodactylus.irc.event.ChannelNotJoined.Reason.registeredNicknamesOnly; import static net.pterodactylus.irc.util.MessageCleaner.getDefaultInstance; import static net.pterodactylus.xdcc.data.Channel.TO_NETWORK; import static net.pterodactylus.xdcc.data.Download.FILTER_RUNNING; @@ -28,18 +32,25 @@ import java.io.IOException; import java.io.OutputStream; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.stream.Collectors; import net.pterodactylus.irc.Connection; -import net.pterodactylus.irc.ConnectionBuilder; +import net.pterodactylus.irc.ConnectionFactory; import net.pterodactylus.irc.DccReceiver; +import net.pterodactylus.irc.DefaultConnection; import net.pterodactylus.irc.event.ChannelJoined; import net.pterodactylus.irc.event.ChannelLeft; import net.pterodactylus.irc.event.ChannelMessageReceived; +import net.pterodactylus.irc.event.ChannelNotJoined; import net.pterodactylus.irc.event.ClientQuit; import net.pterodactylus.irc.event.ConnectionClosed; import net.pterodactylus.irc.event.ConnectionEstablished; @@ -48,6 +59,7 @@ import net.pterodactylus.irc.event.DccAcceptReceived; import net.pterodactylus.irc.event.DccDownloadFailed; import net.pterodactylus.irc.event.DccDownloadFinished; import net.pterodactylus.irc.event.DccSendReceived; +import net.pterodactylus.irc.event.KickedFromChannel; import net.pterodactylus.irc.event.NicknameChanged; import net.pterodactylus.irc.event.PrivateMessageReceived; import net.pterodactylus.irc.event.PrivateNoticeReceived; @@ -63,12 +75,12 @@ import net.pterodactylus.xdcc.core.event.GenericMessage; import net.pterodactylus.xdcc.core.event.MessageReceived; import net.pterodactylus.xdcc.data.Bot; import net.pterodactylus.xdcc.data.Channel; +import net.pterodactylus.xdcc.data.ConnectedNetwork; import net.pterodactylus.xdcc.data.Download; import net.pterodactylus.xdcc.data.Network; import net.pterodactylus.xdcc.data.Pack; import net.pterodactylus.xdcc.data.Server; -import com.google.common.base.Function; import com.google.common.base.Optional; import com.google.common.base.Predicate; import com.google.common.collect.FluentIterable; @@ -98,8 +110,13 @@ public class Core extends AbstractExecutionThreadService { /** The logger. */ private static final Logger logger = Logger.getLogger(Core.class.getName()); + private final Object syncObject = new Object(); /** The event bus. */ private final EventBus eventBus; + private final ConnectionFactory connectionFactory; + private final ChannelBanManager channelBanManager = + new ChannelBanManager(); + private final ConnectionBackoff connectionBackoff = new ConnectionBackoff(); /** The temporary directory to download files to. */ private final String temporaryDirectory; @@ -112,6 +129,7 @@ public class Core extends AbstractExecutionThreadService { /** The channels that are currentlymonitored. */ private final Collection joinedChannels = Sets.newHashSet(); + private final Set channelsBeingJoined = new HashSet<>(); /** The channels that are joined but not configured. */ private final Collection extraChannels = Sets.newHashSet(); @@ -139,8 +157,9 @@ public class Core extends AbstractExecutionThreadService { * The directory to move finished files to */ @Inject - public Core(EventBus eventBus, String temporaryDirectory, String finalDirectory) { + public Core(EventBus eventBus, ConnectionFactory connectionFactory, String temporaryDirectory, String finalDirectory) { this.eventBus = eventBus; + this.connectionFactory = connectionFactory; this.temporaryDirectory = temporaryDirectory; this.finalDirectory = finalDirectory; } @@ -172,13 +191,25 @@ public class Core extends AbstractExecutionThreadService { * * @return All connected networks */ - public Collection connectedNetworks() { - return Lists.newArrayList(Optional.presentInstances(FluentIterable.from(networkConnections.values()).transform(new Function>() { - @Override - public Optional apply(Connection connection) { - return getNetwork(connection); - } - }))); + public Collection connectedNetworks() { + return networkConnections.entrySet().stream().map((entry) -> { + Network network = entry.getKey(); + Collection bots = networkBots.row(network).values(); + int packCount = bots.stream().mapToInt((bot) -> bot.packs().size()).reduce((a, b) -> a + b).orElse(0); + return new ConnectedNetwork(network, entry.getValue().hostname(), + entry.getValue().port(), entry.getValue().nickname(), + channels.stream() + .filter((channel) -> channel.network() + .equals(network)) + .map(Channel::name) + .collect(Collectors.toList()), + extraChannels.stream() + .filter((channel) -> channel.network() + .equals(network)) + .map(Channel::name) + .collect(Collectors.toList()), + bots.size(), packCount); + }).collect(Collectors.toList()); } /** @@ -353,34 +384,82 @@ public class Core extends AbstractExecutionThreadService { @Override protected void run() throws Exception { while (isRunning()) { - try { - Thread.sleep(TimeUnit.MINUTES.toMillis(1)); - } catch (InterruptedException ie1) { - /* ignore. */ - } - /* find channels that should be monitored but are not. */ + Set missingChannels = new HashSet<>(); for (Channel channel : channels) { - if (joinedChannels.contains(channel)) { + if (joinedChannels.contains(channel) || channelsBeingJoined.contains(channel)) { continue; } + if (channelBanManager.isBanned(channel)) { + continue; + } + if (!networkConnections.containsKey(channel.network()) || networkConnections.get(channel.network()).established()) { + missingChannels.add(channel); + } + } + Set missingNetworks = missingChannels.stream() + .map(Channel::network) + .distinct() + .filter((network) -> !networkConnections.containsKey(network)) + .collect(Collectors.toSet()); + + if (missingNetworks.isEmpty()) { + if (!missingChannels.isEmpty()) { + for (Channel missingChannel : missingChannels) { + Network network = missingChannel.network(); + eventBus.post(new GenericMessage(String.format("Trying to join %s on %s...", missingChannel.name(), network))); + try { + channelsBeingJoined.add(missingChannel); + networkConnections.get(network).joinChannel(missingChannel.name()); + } catch (IOException ioe1) { + logger.warn(String.format("Could not join %s on %s!", missingChannel.name(), network.name()), ioe1); + } + } + } else { + synchronized (syncObject) { + try { + syncObject.wait(TimeUnit.MINUTES.toMillis(1)); + } catch (InterruptedException ie1) { + /* ignore. */ + } + } + } + continue; + } + + Map timesForNextConnects = new TreeMap<>(missingNetworks.stream() + .collect(Collectors.toMap(connectionBackoff::getConnectionTime, Function.identity(), (network, ignore) -> network))); - connectNetwork(channel.network()); - Connection connection = networkConnections.get(channel.network()); - if (connection.established()) { - eventBus.post(new GenericMessage(String.format("Trying to join %s on %s.", channel.name(), channel.network().name()))); + Optional> firstNetwork = Optional.fromNullable(timesForNextConnects.entrySet().stream().findFirst().orElse(null)); + if (!firstNetwork.isPresent()) { + continue; + } + if (firstNetwork.get().getKey() > System.currentTimeMillis()) { + eventBus.post(new GenericMessage(String.format("Will connect to %2$s at %1$tH:%1$tM...", firstNetwork.get().getKey(), firstNetwork.get().getValue().name()))); + synchronized (syncObject) { try { - connection.joinChannel(channel.name()); - } catch (IOException ioe1) { - eventBus.post(new GenericMessage(String.format("Could not join %s on %s.", channel.name(), channel.network().name()))); + syncObject.wait(firstNetwork.get().getKey() - System.currentTimeMillis()); + } catch (InterruptedException ie1) { + /* ignore. */ } } + if (!isRunning()) { + break; + } + if (firstNetwork.get().getKey() > System.currentTimeMillis()) { + continue; + } } + + connectNetwork(firstNetwork.get().getValue()); } } @Override - protected void shutDown() { + protected void triggerShutdown() { + synchronized (syncObject) { + syncObject.notifyAll(); + } } // @@ -403,10 +482,12 @@ public class Core extends AbstractExecutionThreadService { return; } Server server = servers.get((int) (Math.random() * servers.size())); - Connection connection = new ConnectionBuilder(eventBus).connect(server.hostname()).port(server.unencryptedPorts().iterator().next()).build(); + eventBus.post(new GenericMessage(String.format("Connecting to %s on %s...", network.name(), server.hostname()))); + Connection connection = connectionFactory.createConnection(server.hostname(), + server.unencryptedPorts().iterator().next()); connection.username(RandomNickname.get()).realName(RandomNickname.get()); networkConnections.put(network, connection); - connection.start(); + connection.open(); } } @@ -421,6 +502,10 @@ public class Core extends AbstractExecutionThreadService { if (!network.isPresent()) { return; } + networkConnections.remove(network.get()); + if (!connection.established()) { + return; + } /* find all channels that need to be removed. */ for (Collection channels : ImmutableList.of(joinedChannels, extraChannels)) { @@ -443,9 +528,6 @@ public class Core extends AbstractExecutionThreadService { } bots.clear(); eventBus.post(new GenericMessage(String.format("Network %s disconnected, %d bots removed, %d packs removed.", network.get().name(), botCount, packCount))); - - /* now remove the network. */ - networkConnections.remove(network.get()); } // @@ -471,6 +553,7 @@ public class Core extends AbstractExecutionThreadService { return; } + connectionBackoff.connectionSuccessful(network.get()); eventBus.post(new GenericMessage(String.format("Connected to network %s.", network.get().name()))); /* join all channels on this network. */ @@ -494,7 +577,12 @@ public class Core extends AbstractExecutionThreadService { */ @Subscribe public void connectionClosed(ConnectionClosed connectionClosed) { + connectionBackoff.connectionFailed(getNetwork(connectionClosed.connection()).get()); removeConnection(connectionClosed.connection()); + synchronized (syncObject) { + syncObject.notifyAll(); + } + eventBus.post(new GenericMessage(String.format("Connection closed by %s.", connectionClosed.connection().hostname()))); } /** @@ -505,7 +593,12 @@ public class Core extends AbstractExecutionThreadService { */ @Subscribe public void connectionFailed(ConnectionFailed connectionFailed) { + connectionBackoff.connectionFailed(getNetwork(connectionFailed.connection()).get()); removeConnection(connectionFailed.connection()); + synchronized (syncObject) { + syncObject.notifyAll(); + } + eventBus.post(new GenericMessage(String.format("Could not connect to %s: %s.", connectionFailed.connection().hostname(), connectionFailed.cause()))); } /** @@ -530,11 +623,60 @@ public class Core extends AbstractExecutionThreadService { return; } + channelBanManager.unban(channel.get()); joinedChannels.add(channel.get()); + channelsBeingJoined.remove(channel.get()); logger.info(String.format("Joined Channel %s on %s.", channelJoined.channel(), network.get().name())); } } + @Subscribe + public void channelNotJoined(ChannelNotJoined channelNotJoined) { + Optional network = getNetwork(channelNotJoined.connection()); + if (!network.isPresent()) { + return; + } + + Optional channel = getChannel(network.get(), channelNotJoined.channel()); + synchronized (syncObject) { + syncObject.notifyAll(); + } + if (!channel.isPresent()) { + eventBus.post(new GenericMessage(format("Could not join %s but didn’t try to join, either.", channelNotJoined.channel()))); + return; + } + channelsBeingJoined.remove(channel.get()); + + + /* remove all bots for this channel, we might have been kicked. */ + Collection botsToRemove = networkBots.row(network.get()) + .values().stream() + .filter(bot -> bot.channel() + .equalsIgnoreCase(channel.get().name())) + .collect(Collectors.toSet()); + botsToRemove.stream() + .forEach(bot -> networkBots.row(network.get()) + .remove(bot.name())); + + channelBanManager.ban(channel.get()); + if (channelNotJoined.reason() == registeredNicknamesOnly) { + eventBus.post(new GenericMessage( + format("Not trying to join %s anymore.", channel.get()))); + } else if (channelNotJoined.reason() == inviteOnly) { + eventBus.post(new GenericMessage( + format("%s is invite-only, suspending join for a day.", + channel.get()))); + } else if (channelNotJoined.reason() == banned) { + eventBus.post(new GenericMessage( + format("Banned from %s, suspending join for a day.", + channel.get()))); + } else { + eventBus.post(new GenericMessage( + format("Could not join %s: %s", channelNotJoined.channel(), + channelNotJoined.reason()))); + } + } + /** * Removes bots that leave a channel, or channels when it’s us that’s leaving. * @@ -565,6 +707,9 @@ public class Core extends AbstractExecutionThreadService { } else { channels.remove(channel.get()); } + synchronized (syncObject) { + syncObject.notifyAll(); + } eventBus.post(new GenericMessage(String.format("Left Channel %s on %s.", channel.get().name(), channel.get().network().name()))); } @@ -575,6 +720,44 @@ public class Core extends AbstractExecutionThreadService { networkBots.remove(network.get(), channelLeft.client().nick().get()); } + @Subscribe + public void kickedFromChannel(KickedFromChannel kickedFromChannel) { + Optional network = getNetwork(kickedFromChannel.connection()); + if (!network.isPresent()) { + return; + } + + /* have we been kicked? */ + if (nicknameMatchesConnection(kickedFromChannel.connection(), kickedFromChannel.kickee())) { + Optional channel = getChannel(network.get(), kickedFromChannel.channel()); + if (!channel.isPresent()) { + /* maybe it was an extra channel? */ + channel = getExtraChannel(network.get(), kickedFromChannel.channel()); + if (!channel.isPresent()) { + /* okay, whatever. */ + return; + } + + extraChannels.remove(channel.get()); + } else { + joinedChannels.remove(channel.get()); + } + synchronized (syncObject) { + syncObject.notifyAll(); + } + eventBus.post(new GenericMessage(format( + "Kicked from %s by %s: %s", + kickedFromChannel.channel(), + kickedFromChannel.kicker(), + kickedFromChannel.reason().or("") + ))); + } + } + + private boolean nicknameMatchesConnection(Connection connection, String nickname) { + return connection.nickname().equalsIgnoreCase(nickname); + } + /** * Removes a client (which may be a bot) from the table of known bots. * @@ -643,7 +826,8 @@ public class Core extends AbstractExecutionThreadService { Bot bot; synchronized (networkBots) { if (!networkBots.contains(network.get(), channelMessageReceived.source().nick().get())) { - bot = new Bot(network.get(), channelMessageReceived.source().nick().get()); + bot = new Bot(network.get(), channelMessageReceived.channel(), + channelMessageReceived.source().nick().get()); networkBots.put(network.get(), channelMessageReceived.source().nick().get(), bot); eventBus.post(new BotAdded(bot)); } else { @@ -680,7 +864,7 @@ public class Core extends AbstractExecutionThreadService { return; } - eventBus.post(new GenericMessage(String.format("Notice from %s (%s): %s", privateNoticeReceived.reply().source().get(), network.get(), privateNoticeReceived.text()))); + eventBus.post(new GenericMessage(String.format("Notice from %s (%s): %s", privateNoticeReceived.source(), network.get(), privateNoticeReceived.text()))); } /**