X-Git-Url: https://git.pterodactylus.net/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Fxdcc%2Fcore%2FCore.java;h=00d1fbdf4a949ffd78da8d3873ce6351601f3890;hb=44825f1cf5ddb021cbf9b8416e8649e768905e77;hp=6da5dbdcd57152b4f2152188694debde474ea5be;hpb=60eeb8cc8d2ccd56dfae68cea32012443bbe50a8;p=xudocci.git diff --git a/src/main/java/net/pterodactylus/xdcc/core/Core.java b/src/main/java/net/pterodactylus/xdcc/core/Core.java index 6da5dbd..00d1fbd 100644 --- a/src/main/java/net/pterodactylus/xdcc/core/Core.java +++ b/src/main/java/net/pterodactylus/xdcc/core/Core.java @@ -19,6 +19,7 @@ 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; @@ -31,11 +32,15 @@ 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; @@ -111,6 +116,7 @@ public class Core extends AbstractExecutionThreadService { 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; @@ -123,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(); @@ -377,39 +384,74 @@ public class Core extends AbstractExecutionThreadService { @Override protected void run() throws Exception { while (isRunning()) { - synchronized (syncObject) { - try { - syncObject.wait(TimeUnit.MINUTES.toMillis(1)); - } catch (InterruptedException ie1) { - /* ignore. */ - } - } - if (!isRunning()) { - break; - } - /* 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; } - - /* are we banned from this channel? */ 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; + } - 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()))); + Map timesForNextConnects = new TreeMap<>(missingNetworks.stream() + .collect(Collectors.toMap(connectionBackoff::getConnectionTime, Function.identity(), (network, ignore) -> network))); + + 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("Waiting %d minutes to connect to %s...", TimeUnit.MILLISECONDS.toMinutes(firstNetwork.get().getKey() - System.currentTimeMillis()), 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()); } } @@ -440,6 +482,7 @@ public class Core extends AbstractExecutionThreadService { return; } Server server = servers.get((int) (Math.random() * servers.size())); + 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()); @@ -510,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. */ @@ -533,7 +577,11 @@ 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()))); } @@ -545,7 +593,11 @@ 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()))); } @@ -573,6 +625,7 @@ public class Core extends AbstractExecutionThreadService { 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())); } } @@ -585,10 +638,15 @@ public class Core extends AbstractExecutionThreadService { } 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.", channel.get()))); + 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()) @@ -606,6 +664,13 @@ public class Core extends AbstractExecutionThreadService { format("Not trying to join %s anymore.", channel.get()))); return; } + if (channelNotJoined.reason() == inviteOnly) { + channelBanManager.ban(channel.get()); + eventBus.post(new GenericMessage( + format("%s is invite-only, suspending join for a day.", + channel.get()))); + return; + } if (channelNotJoined.reason() == banned) { channelBanManager.ban(channel.get()); eventBus.post(new GenericMessage( @@ -649,6 +714,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()))); } @@ -681,6 +749,9 @@ public class Core extends AbstractExecutionThreadService { } else { joinedChannels.remove(channel.get()); } + synchronized (syncObject) { + syncObject.notifyAll(); + } eventBus.post(new GenericMessage(format( "Kicked from %s by %s: %s", kickedFromChannel.channel(),