/// <summary>Parse a revocation message.</summary>
    public UserRevocationMessage(Certificate cacert, MemBlock data)
    {
      _data = data;

      int pos = 0;
      int length = 0;

      Username = AdrConverter.Deserialize(data, pos, out length) as string;
      pos += length;
      // Random number to reduce likelihood of malicious duplication of messages
      NumberSerializer.ReadInt(data, pos);
      pos += 4;
      // Verify that there is a date contained therein, perhaps we should verify the time
      new DateTime(NumberSerializer.ReadLong(data, pos));
      pos += 8;
      Signature = new byte[data.Length - pos];
      data.Slice(pos).CopyTo(Signature, 0);

      // hash the data
      SHA1CryptoServiceProvider sha1 = new SHA1CryptoServiceProvider();
      Hash = sha1.ComputeHash(data, 0, data.Length - Signature.Length);

      if(!cacert.PublicKey.VerifyHash(Hash,
            CryptoConfig.MapNameToOID("SHA1"), Signature))
      {
        throw new Exception("Invalid UserRevocationMessage signature");
      }
    }

 //Too bad we can't use a template here, .Net generics *may* do the job
 public static int ReadInt(MemBlock mb, int offset)
 {
   int val = 0;
   for(int i = 0; i < 4; i++) {
     val = (val << 8) | mb[i + offset];
   }
   return val;
 }

 public DhtGet(Node node, MemBlock key, EventHandler enqueue,
     EventHandler close) : base(close)
 {
   Node = node;
   Key = key;
   _enqueue = enqueue;
   Results = new Queue<MemBlock>();
 }

 public void HandleData(MemBlock b, ISender return_path, object state) {
   byte b0 = b[0];
   if( b0 == 0 ) {
     //This is a request:
     MemBlock data = b.Slice(1);
     //Make sure node to reply with a zero
     return_path.Send( new CopyList( PType.Protocol.Echo, REPLY_HEADER, data) );
   }
 }

 public DhtPut(Node node, MemBlock key, MemBlock value, int ttl,
     EventHandler finished) : base(finished)
 {
   Node = node;
   Key = key;
   Value = value;
   Ttl = ttl;
   _result = null;
 }

 override public void HandleData(MemBlock data, ISender return_path, object state)
 {
   SecurityAssociation sa = return_path as SecurityAssociation;
   if(sa == null) {
     ProtocolLog.WriteIf(ProtocolLog.Exceptions, String.Format(
           "Insecure sender {0} sent ptype {1}", return_path, _ptype));
     return;
   }
   base.HandleData(data, return_path, state);
 }

 /// <summary>Create a new Attribute.</summary>
 public Attribute(AttributeType type, MemBlock value)
 {
   Type = type;
   Value = value;
   byte[] data = new byte[4 + value.Length];
   NumberSerializer.WriteUShort((ushort) type, data, 0);
   NumberSerializer.WriteUShort((ushort) value.Length, data, 2);
   value.CopyTo(data, 4);
   Data = MemBlock.Reference(data);
 }

 /**
  * This handles the packet forwarding protocol
  */
 public void HandleData(MemBlock b, ISender ret_path, object state)
 {
   /*
    * Check it
    */
   AHSender ahs = ret_path as AHSender;
   if( ahs != null ) {
     //This was an AHSender:
     /*
      * This goes A -> B -> C
      */
     if( b[0] == 0 ) {
       int offset = 1;
       //This is the first leg, going from A->B
       Address add_c = AddressParser.Parse(b.Slice(offset, Address.MemSize));
       offset += Address.MemSize;
       //Since ahs a sender to return, we would be the source:
       Address add_a = ahs.Destination;
       short ttl = NumberSerializer.ReadShort(b, offset);//2 bytes
       offset += 2;
       ushort options = (ushort) NumberSerializer.ReadShort(b, offset);//2 bytes
       offset += 2;
       MemBlock payload = b.Slice(offset);
       MemBlock f_header = MemBlock.Reference( new byte[]{1} );
       /*
        * switch the packet from [A B f0 C] to [B C f 1 A]
        */
       ICopyable new_payload = new CopyList(PType.Protocol.Forwarding,
                                        f_header, add_a, payload);
       /*
        * ttl and options are present in the forwarding header.
        */
       AHSender next = new AHSender(_n, ahs.ReceivedFrom, add_c,
                                    ttl,
                                    options); 
       next.Send(new_payload);
     }
     else if ( b[0] == 1 ) {
       /*
        * This is the second leg: B->C
        * Make a Forwarding Sender, and unwrap the inside packet
        */
       Address add_a = AddressParser.Parse(b.Slice(1, Address.MemSize));
       Address add_b = ahs.Destination;
       MemBlock rest_of_payload = b.Slice(1 + Address.MemSize);
       //Here's the return path:
       ISender new_ret_path = new ForwardingSender(_n, add_b, add_a);
       _n.HandleData(rest_of_payload, new_ret_path, this);
     }
   }
   else {
     //This is not (currently) supported.
     Console.Error.WriteLine("Got a forwarding request from: {0}", ret_path);
   }
 }

 public DirectionalAddress(MemBlock mb)
 {
   if (ClassOf(mb) != this.Class) {
     throw new System.
     ArgumentException
     ("This is not an AHAddress (Class 124) :  ",
      this.ToString());
   }
   _buffer = mb;
   _dir = (Direction) NumberSerializer.ReadInt(mb, 0);
 }

 public void HandleData(MemBlock data, ISender return_path, object state)
 {
   MemBlock user_data;
   // Parse BroadcastSender
   BroadcastSender bs = BroadcastSender.Parse(Node, data, out user_data);
   // Present the packet to the local handler
   BroadcastReceiver br = new BroadcastReceiver(bs);
   Node.HandleData(user_data, br, null);
   // Broadcast to the next hop
   bs.Send(user_data);
 }

    public ConnectionHandler(PType ptype, StructuredNode node)
    {
      _node = node;
      _ondemand = new OnDemandConnectionOverlord(node);
      _node.AddConnectionOverlord(_ondemand);
      _ptype = ptype;
      _ptype_mb = ptype.ToMemBlock();
      _address_to_sender = new Dictionary<Address, ISender>();
      _sender_to_address = new Dictionary<ISender, Address>();

      node.GetTypeSource(_ptype).Subscribe(this, null);
      node.ConnectionTable.ConnectionEvent += HandleConnection;
      node.ConnectionTable.DisconnectionEvent += HandleDisconnection;
    }

    /// <summary>Uses the Dht for the bootstrap problem.</summary>
    /// <param name="node">The node needing remote tas.</param>
    /// <param name="dht">The dht for the shared overlay.</param>
    /// <param name="dht_proxy">A dht proxy for the shared overlay.</param>
    public DhtDiscovery(StructuredNode node, IDht dht, string shared_namespace,
        RpcDhtProxy dht_proxy) :
      base(node)
    {
      _dht = dht;
      _dht_proxy = dht_proxy;
      _node = node;
      _shared_namespace = shared_namespace;
      string skey = "PrivateOverlay:" + node.Realm;
      byte[] bkey = Encoding.UTF8.GetBytes(skey);
      _p2p_address = node.Address.ToMemBlock();
      _private_dht_key = MemBlock.Reference(bkey);

      _ongoing = 0;
      _steady_state = 0;
      _dht_proxy.Register(_private_dht_key, _p2p_address, PUT_DELAY_S);
    }

 public void HandleData(MemBlock b, ISender from, object state) {
   MemBlock payload = null;
   PType t = null;
   try {
     t = PType.Parse(b, out payload);
     if(t.Equals(PType.Protocol.ReqRep)) {
       _rrm.HandleData(payload, from, state);
     }
     else if(t.Equals(PType.Protocol.Rpc)) {
       Rpc.HandleData(payload, from, state);
     }
   }
   catch(Exception x) {
     Console.Error.WriteLine("Packet Handling Exception: {3}\n\tType: {0}\n\t\n\tFrom: {1}\n\tData: {2}",
       t, from, payload.GetString(System.Text.Encoding.ASCII), x);
   }
 }

    public GroupGraph(int count, int near, int shortcuts, int random_seed,
        List<List<int>> dataset, int group_count) :
      base(count, near, shortcuts, random_seed, dataset)
    {
      if(group_count > count || group_count < 0) {
        throw new Exception("Invalid group count: " + group_count);
      }

      _group_members = new List<GraphNode>();

      for(int i = 0; i < group_count; i++) {
        int index = _rand.Next(0, count);
        AHAddress addr = _addrs[index];
        _group_members.Add(_addr_to_node[addr]);
      }

      _group_identifier = GenerateAddress().ToMemBlock();
    }

    /// <summary>Parses an incoming revocation and updates the revoked users
    /// hashtable if successful.</summary>
    public void HandleData(MemBlock data, ISender ret, object state)
    {
      UserRevocationMessage urm = null;
      try {
        urm = new UserRevocationMessage(_ca_cert, data);
      } catch(Exception e) {
        ProtocolLog.WriteIf(ProtocolLog.SecurityExceptions, e.ToString());
        return;
      }


      lock(_revoked_users) {
        if(_revoked_users.Contains(urm.Username)) {
          return;
        }
        _revoked_users[urm.Username] = true;
      }

      _so.VerifySAs();
    }

  /**
   * Here we handle routing AHPackets
   */
  public void HandleData(MemBlock data, ISender ret_path, object st) {
    AHState state = _state; //Read the state, it can't change after the read
    var header = new AHHeader(data);
    var payload = data.Slice(header.Length);

    Connection next_con;
    //Check to see if we can use a Leaf connection:
    int dest_idx = state.Leafs.IndexOf(header.Destination);
    if( dest_idx >= 0 ) {
      next_con = state.Leafs[dest_idx];
    }
    else {
      var alg = state.GetRoutingAlgo(header);
      Pair<Connection, bool> result = alg.NextConnection(ret_path as Edge, header);
      if( result.Second ) {
        //Send a response exactly back to the node that sent to us
        var resp_send = new AHSender(_n, ret_path, header.Source,
                                       AHSender.DefaultTTLFor(_n.NetworkSize),
                                       AHHeader.Options.Exact);
        _n.HandleData( payload, resp_send, this); 
      }
      next_con = result.First;
    }
    //Send it on:
    if( next_con != null ) {
      //Now we do the sending:
      var new_packet = new CopyList(PType.Protocol.AH,
                                    header.IncrementHops(),
                                    payload);
      try {
        next_con.Edge.Send(new_packet);
      }
      catch(EdgeException) {
        //Just drop the packet...
      }
    }
  }

  public void Send(ICopyable data) {
    /*
     * Assemble an AHPacket:
     */
    if( _header == null ) {
      AHHeader ahh = new AHHeader(_hops, _ttl, _source, _dest, _options);
      _header = MemBlock.Copy(new CopyList( PType.Protocol.AH, ahh));
      _header_length = _header.Length;
    }
    byte[] ah_packet;
    int packet_length;
    int packet_offset;

    //Try to get the shared BufferAllocator, useful when
    //we don't know how big the data is, which in general
    //is just as expensive as doing a CopyTo...
    BufferAllocator ba = Interlocked.Exchange<BufferAllocator>(ref _buf_alloc, null);
    if( ba != null ) {
      try {
        ah_packet = ba.Buffer;
        packet_offset = ba.Offset;
        int tmp_off = packet_offset;
        tmp_off += _header.CopyTo(ah_packet, packet_offset);
        tmp_off += data.CopyTo(ah_packet, tmp_off);
        packet_length = tmp_off - packet_offset;
        ba.AdvanceBuffer(packet_length);
      }
      catch(System.Exception x) {
        throw new SendException(false, "could not write the packet, is it too big?", x);
      }
      finally {
        //Put the BA back
        Interlocked.Exchange<BufferAllocator>(ref _buf_alloc, ba);
      }
    }
    else {
      //Oh well, someone else is using the buffer, just go ahead
      //and allocate new memory:
      packet_offset = 0;
      packet_length = _header_length + data.Length;
      ah_packet = new byte[ packet_length ];
      int off_to_data = _header.CopyTo(ah_packet, 0);
      data.CopyTo(ah_packet, off_to_data);
    }
    MemBlock mb_packet = MemBlock.Reference(ah_packet, packet_offset, packet_length);
    /*
     * Now we announce this packet, the AHHandler will
     * handle routing it for us
     */
    _n.HandleData(mb_packet, _from, this);
  }

 public void HandleData(MemBlock data, ISender sender, object state)
 {
   BroadcastReceiver br = sender as BroadcastReceiver; 
   _estimated_time_left = DateTime.UtcNow.AddSeconds(1);
   _results.Add(br);
 }

    /** Handle incoming data on an Edge 
     */
    public void HandleData(MemBlock data, ISender retpath, object state) {
      MemBlock rest_of_data;
      PType p;

      if( state == null ) {
        try {
          p = PType.Parse(data, out rest_of_data);
        } catch(ParseException) {
          ProtocolLog.WriteIf(ProtocolLog.Pathing, "Invalid PType from: " + data);
          return;
        }
        p = PType.Parse(data, out rest_of_data);
      }
      else {
        //a demux has already happened:
        p = (PType)state;
        rest_of_data = data;
      }

      if( PType.Protocol.Pathing.Equals(p) ) {
        /*
         * We use a special PType to denote this transaction so
         * we don't confuse it with other RepRep communication
         */
        _rrm.HandleData(rest_of_data, retpath, null);
      }
      else if( PType.Protocol.Rpc.Equals(p) ) {
       /*
        * Send this to the RpcHandler
        */
       Rpc.HandleData(rest_of_data, retpath, null);
      }
      else {
        /*
         * This is some other data
         * It is either:
         * 1) Time to announce an already created edge.
         * 2) Assume this is a "default path" edge creation, to be backwards
         * compatible
         */
        Edge e = null;
        PathEdge pe = null;
        try {
          e = (Edge)retpath;
          PathEdgeListener pel = null;
          lock( _sync ) {
            if( _unannounced.TryGetValue(e, out pe) ) {
              //
              _unannounced.Remove(e);
              pel = _pel_map[pe.LocalPath];
            }
          }
          if( pe == null ) {
            if(! _pel_map.ContainsKey(Root) ) {
              ProtocolLog.WriteIf(ProtocolLog.Pathing, "No root, can't create edge");
              if(e != null) {
                e.Close();
              }
              return;
            }
            /*
             * This must be a "default path" incoming connection
             */
            pel = _pel_map[Root];
            pe = new PathEdge(this, e, Root, Root);
          }
          pel.SendPathEdgeEvent(pe);
          pe.Subscribe();
          pe.ReceivedPacketEvent(data);
        }
        catch(Exception x) {
          if( pe != null ) {
            //This closes both edges:
            pe.Close();  
          }
          else if( e != null ) {
            ProtocolLog.WriteIf(ProtocolLog.Pathing,
                String.Format("Closing ({0}) due to: {1}", e, x));
            e.Close();  
          }
        }
      }
    }

    /**
     * This reads a packet from buf which came from end, with
     * the given ids
     */
    protected void HandleDataPacket(int remoteid, int localid,
                                    MemBlock packet, EndPoint end, object state)
    {
      bool read_packet = true;
      bool is_new_edge = false;
      //It is threadsafe to read from Hashtable
      UdpEdge edge = (UdpEdge)_id_ht[localid];
      if( localid == 0 ) {
        //This is a potentially a new incoming edge
        is_new_edge = true;

        //Check to see if it is a dup:
        UdpEdge e_dup = (UdpEdge)_remote_id_ht[remoteid];
        if( e_dup != null ) {
          //Lets check to see if this is a true dup:
          if( e_dup.End.Equals( end ) ) {
            //Same id from the same endpoint, looks like a dup...
            is_new_edge = false;
            //Reuse the existing edge:
            edge = e_dup;
          }
          else {
            //This is just a coincidence.
          }
        }
        if( is_new_edge ) {
          TransportAddress rta = TransportAddressFactory.CreateInstance(this.TAType,(IPEndPoint)end);
          if( _ta_auth.Authorize(rta) == TAAuthorizer.Decision.Deny ) {
            //This is bad news... Ignore it...
            ///@todo perhaps we should send a control message... I don't know
            is_new_edge= false;
            read_packet = false;
            if(ProtocolLog.UdpEdge.Enabled)
              ProtocolLog.Write(ProtocolLog.UdpEdge, String.Format(
                "Denying: {0}", rta));
          }
          else {
            //We need to assign it a local ID:
            lock( _id_ht ) {
              /*
               * Now we need to lock the table so that it cannot
               * be written to by anyone else while we work
               */
              do {
                localid = _rand.Next();
                //Make sure not to use negative ids
                if( localid < 0 ) { localid = ~localid; }
              } while( _id_ht.Contains(localid) || localid == 0 );
              /*
               * We copy the endpoint because (I think) .Net
               * overwrites it each time.  Since making new
               * edges is rare, this is better than allocating
               * a new endpoint each time
               */
              IPEndPoint this_end = (IPEndPoint)end;
              IPEndPoint my_end = new IPEndPoint(this_end.Address,
                                                 this_end.Port);
              edge = new UdpEdge(_send_handler, true, my_end,
                             _local_ep, localid, remoteid);
              _id_ht[localid] = edge;
              _remote_id_ht[remoteid] = edge;
            }
          }
        }
      }
      else if ( edge == null ) {
        /*
         * This is the case where the Edge is not a new edge,
         * but we don't know about it.  It is probably an old edge
         * that we have closed.  We can ignore this packet
         */
        read_packet = false;
         //Send a control packet
        SendControlPacket(end, remoteid, localid, ControlCode.EdgeClosed, state);
      }
      else if ( edge.RemoteID == 0 ) {
        /* This is the response to our edge creation */
        edge.RemoteID = remoteid;
      }
      else if( edge.RemoteID != remoteid ) {
        /*
         * This could happen as a result of packet loss or duplication
         * on the first packet.  We should ignore any packet that
         * does not have both ids matching.
         */
        read_packet = false;
         //Tell the other guy to close this ignored edge
        SendControlPacket(end, remoteid, localid, ControlCode.EdgeClosed, state);
        edge = null;
      }
      if( (edge != null) && !edge.End.Equals(end) ) {
        //This happens when a NAT mapping changes
        if(ProtocolLog.UdpEdge.Enabled)
          ProtocolLog.Write(ProtocolLog.UdpEdge, String.Format(
            "Remote NAT Mapping changed on Edge: {0}\n{1} -> {2}",
            edge, edge.End, end)); 
//.........这里部分代码省略.........