genericmessageconnectionindirect.java

java 文件下载器。可自定义

				} else
				{
					Debug.out((new StringBuilder()).append("Unexpected reply type - ").append(reply_type).toString());
					listener.connectFailure(new Throwable((new StringBuilder()).append("Unexpected reply type - ").append(reply_type).toString()));
				}
			}
		}
		catch (Throwable e)
		{
			listener.connectFailure(e);
		}
	}

	public void accepted()
	{
	}

	public void send(PooledByteBuffer pbb)
		throws MessageException
	{
		byte bytes[] = pbb.toByteArray();
		if (incoming)
		{
			synchronized (send_queue)
			{
				if (send_queue.size() > 64)
					throw new MessageException("Send queue limit exceeded");
				send_queue.add(bytes);
			}
			send_queue_sem.release();
		} else
		{
			List messages = new ArrayList();
			messages.add(bytes);
			send(messages);
		}
	}

	protected void send(List messages)
	{
		try
		{
			Map message = new HashMap();
			message.put("con_id", new Long(connection_id));
			message.put("type", new Long(3L));
			message.put("data", messages);
			Map reply = nat_traverser.sendMessage(message_manager, rendezvous, target, message);
			last_message_sent = SystemTime.getCurrentTime();
			if (reply == null || !reply.containsKey("type"))
			{
				owner.reportFailed(new Throwable((new StringBuilder()).append("Indirect message send failed (response=").append(reply).append(")").toString()));
			} else
			{
				int reply_type = ((Long)reply.get("type")).intValue();
				if (reply_type == 2)
					owner.reportFailed(new Throwable(new String((byte[])(byte[])reply.get("error"))));
				else
				if (reply_type == 3)
				{
					List replies = (List)reply.get("data");
					for (int i = 0; i < replies.size(); i++)
						owner.receive(new GenericMessage(msg_id, msg_desc, new DirectByteBuffer(ByteBuffer.wrap((byte[])replies.get(i))), false));

					if (reply.get("more_data") != null)
						new DelayedEvent("GenMsg:kap", 500L, new AERunnable() {

							final GenericMessageConnectionIndirect this$0;

							public void runSupport()
							{
								if (prepareForKeepAlive(true))
									GenericMessageConnectionIndirect.keep_alive_pool.run(new AERunnable() {

										final 2 this$1;

										public void runSupport()
										{
											keepAlive();
										}

					
					{
						this$1 = 2.this;
						super();
					}
									});
							}

			
			{
				this$0 = GenericMessageConnectionIndirect.this;
				super();
			}
						});
				} else
				if (reply_type == 4)
					owner.reportFailed(new Throwable("Disconnected"));
			}
		}
		catch (Throwable e)
		{
			owner.reportFailed(e);
		}
	}

	protected List receive(List messages)
	{
		last_message_received = SystemTime.getCurrentTime();
		for (int i = 0; i < messages.size(); i++)
			owner.receive(new GenericMessage(msg_id, msg_desc, new DirectByteBuffer(ByteBuffer.wrap((byte[])messages.get(i))), false));

		List reply = new ArrayList();
		if (send_queue_sem.reserve(2500L))
		{
			try
			{
				Thread.sleep(250L);
			}
			catch (Throwable e) { }
			int max = getMaximumMessageSize();
			int total = 0;
			synchronized (send_queue)
			{
				do
				{
					if (send_queue.size() <= 0)
						break;
					byte data[] = (byte[])send_queue.getFirst();
					if (total > 0 && total + data.length > max)
						break;
					reply.add(send_queue.removeFirst());
					total += data.length;
				} while (true);
			}
			if (reply.size() == 0)
			{
				send_queue_sem.release();
			} else
			{
				for (int i = 1; i < reply.size(); i++)
					send_queue_sem.reserve();

			}
		}
		return reply;
	}

	protected boolean receiveIncomplete()
	{
		LinkedList linkedlist = send_queue;
		JVM INSTR monitorenter ;
		return send_queue.size() > 0;
		Exception exception;
		exception;
		throw exception;
	}

	public void close()
		throws MessageException
	{
		close(null);
	}

	protected void close(Throwable close_cause)
		throws MessageException
	{
		if (closed)
			return;
		log((new StringBuilder()).append("connection to ").append(endpoint.getNotionalAddress()).append(" closed").append(close_cause != null ? (new StringBuilder()).append(" (").append(close_cause).append(")").toString() : "").toString());
		closed = true;
		if (incoming)
		{
			synchronized (remote_connections)
			{
				remote_connections.remove(new Long(connection_id));
			}
		} else
		{
			synchronized (local_connections)
			{
				local_connections.remove(new Long(connection_id));
			}
			Map message = new HashMap();
			message.put("con_id", new Long(connection_id));
			message.put("type", new Long(4L));
			try
			{
				nat_traverser.sendMessage(message_manager, rendezvous, target, message);
				last_message_sent = SystemTime.getCurrentTime();
			}
			catch (Throwable e)
			{
				throw new MessageException("Close operation failed", e);
			}
		}
		if (close_cause != null)
			owner.reportFailed(close_cause);
		break MISSING_BLOCK_LABEL_290;
		Exception exception2;
		exception2;
		if (close_cause != null)
			owner.reportFailed(close_cause);
		throw exception2;
	}

	protected boolean isClosed()
	{
		return closed;
	}

	protected boolean prepareForKeepAlive(boolean force)
	{
		if (keep_alive_in_progress)
			return false;
		long now = SystemTime.getCurrentTime();
		if (force || now < last_message_sent || now - last_message_sent > 10000L)
		{
			keep_alive_in_progress = true;
			return true;
		} else
		{
			return false;
		}
	}

	protected void keepAlive()
	{
		send(new ArrayList());
		keep_alive_in_progress = false;
		break MISSING_BLOCK_LABEL_27;
		Exception exception;
		exception;
		keep_alive_in_progress = false;
		throw exception;
	}

	protected static void log(String str)
	{
		if (Logger.isEnabled())
			Logger.log(new LogEvent(LOGID, (new StringBuilder()).append("GenericMessaging (indirect):").append(str).toString()));
	}

	protected void trace(String s)
	{
	}

	static 
	{
		LOGID = LogIDs.NET;
		SimpleTimer.addPeriodicEvent("DDBTorrent:timeout2", 5000L, new TimerEventPerformer() {

			private int tick_count;

			public void perform(TimerEvent event)
			{
				tick_count++;
				if (tick_count % 12 == 0 && Logger.isEnabled())
				{
					int local_total;
					synchronized (GenericMessageConnectionIndirect.local_connections)
					{
						local_total = GenericMessageConnectionIndirect.local_connections.size();
					}
					int remote_total;
					synchronized (GenericMessageConnectionIndirect.remote_connections)
					{
						remote_total = GenericMessageConnectionIndirect.remote_connections.size();
					}
					if (local_total + remote_total > 0)
						GenericMessageConnectionIndirect.log((new StringBuilder()).append("local=").append(local_total).append(" [").append(GenericMessageConnectionIndirect.getLocalConnectionStatus()).append("], remote=").append(remote_total).append(" [").append(GenericMessageConnectionIndirect.getRemoteConnectionStatus()).append("]").toString());
				}
				if (tick_count % 1 == 0)
				{
					synchronized (GenericMessageConnectionIndirect.local_connections)
					{
						Iterator it = GenericMessageConnectionIndirect.local_connections.values().iterator();
						do
						{
							if (!it.hasNext())
								break;
							GenericMessageConnectionIndirect con = (GenericMessageConnectionIndirect)it.next();
							if (con.prepareForKeepAlive(false))
								GenericMessageConnectionIndirect.keep_alive_pool.run(con. new AERunnable() {

									final GenericMessageConnectionIndirect val$con;
									final 1 this$0;

									public void runSupport()
									{
										con.keepAlive();
									}


// JavaClassFileOutputException: Invalid index accessing method local variables table of <init>
								});
						} while (true);
					}
					long now = SystemTime.getCurrentTime();
					synchronized (GenericMessageConnectionIndirect.remote_connections)
					{
						if (GenericMessageConnectionIndirect.remote_connections.size() > 0)
						{
							Iterator it = (new ArrayList(GenericMessageConnectionIndirect.remote_connections.values())).iterator();
							do
							{
								if (!it.hasNext())
									break;
								GenericMessageConnectionIndirect con = (GenericMessageConnectionIndirect)it.next();
								long last_receive = con.getLastMessageReceivedTime();
								if (now - last_receive > 30000L)
									try
									{
										con.close(new Throwable("Timeout"));
									}
									catch (Throwable e)
									{
										Debug.printStackTrace(e);
									}
							} while (true);
						}
					}
				}
			}

			
			{
				tick_count = 0;
			}
		});
	}



}