groupsock.cpp

流媒体传输协议的实现代码,非常有用.可以支持rtsp mms等流媒体传输协议

  if (!wasLoopedBackFromUs(env(), fromAddress)) {
    statsIncoming.countPacket(numBytes);
    statsGroupIncoming.countPacket(numBytes);
    numMembers =
      outputToAllMembersExcept(NULL, ttl(),
			       buffer, bytesRead,
			       fromAddress.sin_addr.s_addr);
    if (numMembers > 0) {
      statsRelayedIncoming.countPacket(numBytes);
      statsGroupRelayedIncoming.countPacket(numBytes);
    }
  }
  if (DebugLevel >= 3) {
    env() << *this << ": read " << bytesRead << " bytes from ";
    env() << our_inet_ntoa(fromAddress.sin_addr);
    if (numMembers > 0) {
      env() << "; relayed to " << numMembers << " members";
    }
    env() << "\n";
  }
  
  return True;
}

Boolean Groupsock::wasLoopedBackFromUs(UsageEnvironment& env,
				       struct sockaddr_in& fromAddress) {
  if (fromAddress.sin_addr.s_addr
      == ourSourceAddressForMulticast(env)) {
    if (fromAddress.sin_port == sourcePortNum()) {
#ifdef DEBUG_LOOPBACK_CHECKING
      if (DebugLevel >= 3) {
	env() << *this << ": got looped-back packet\n";
      }
#endif
      return True;
    }
  }
  
  return False;
}

int Groupsock::outputToAllMembersExcept(DirectedNetInterface* exceptInterface,
					u_int8_t ttlToFwd,
					unsigned char* data, unsigned size,
					netAddressBits sourceAddr) {
  // Don't forward TTL-0 packets
  if (ttlToFwd == 0) return 0;
  
  DirectedNetInterfaceSet::Iterator iter(members());
  unsigned numMembers = 0;
  DirectedNetInterface* interf;
  while ((interf = iter.next()) != NULL) {
    // Check whether we've asked to exclude this interface:
    if (interf == exceptInterface)
      continue;
    
    // Check that the packet's source address makes it OK to
    // be relayed across this interface:
    UsageEnvironment& saveEnv = env();
    // because the following call may delete "this"
    if (!interf->SourceAddrOKForRelaying(saveEnv, sourceAddr)) {
      if (strcmp(saveEnv.getResultMsg(), "") != 0) {
				// Treat this as a fatal error
	return -1;
      } else {
	continue;
      }
    }
    
    if (numMembers == 0) {
      // We know that we're going to forward to at least one
      // member, so fill in the tunnel encapsulation trailer.
      // (Note: Allow for it not being 4-byte-aligned.)
      TunnelEncapsulationTrailer* trailerInPacket
	= (TunnelEncapsulationTrailer*)&data[size];
      TunnelEncapsulationTrailer* trailer;
      
      Boolean misaligned = ((unsigned long)trailerInPacket & 3) != 0;
      unsigned trailerOffset;
      u_int8_t tunnelCmd;
      if (isSSM()) {
	// add an 'auxilliary address' before the trailer
	trailerOffset = TunnelEncapsulationTrailerAuxSize;
	tunnelCmd = TunnelDataAuxCmd;
      } else {
	trailerOffset = 0;
	tunnelCmd = TunnelDataCmd;
      }
      unsigned trailerSize = TunnelEncapsulationTrailerSize + trailerOffset;
      unsigned tmpTr[TunnelEncapsulationTrailerMaxSize];
      if (misaligned) {
	trailer = (TunnelEncapsulationTrailer*)&tmpTr;
      } else {
	trailer = trailerInPacket;
      }
      trailer += trailerOffset;
      
      if (fDests != NULL) {
	trailer->address() = fDests->fGroupEId.groupAddress().s_addr;
	trailer->port() = fDests->fPort; // structure copy, outputs in network order
      }
      trailer->ttl() = ttlToFwd;
      trailer->command() = tunnelCmd;
      
      if (isSSM()) {
	trailer->auxAddress() = sourceFilterAddress().s_addr;
      }
      
      if (misaligned) {
	memmove(trailerInPacket, trailer-trailerOffset, trailerSize);
      }
      
      size += trailerSize;
    }
    
    interf->write(data, size);
    ++numMembers;
  }
  
  return numMembers;
}

UsageEnvironment& operator<<(UsageEnvironment& s, const Groupsock& g) {
  UsageEnvironment& s1 = s << timestampString() << " Groupsock("
			   << g.socketNum() << ": "
			   << our_inet_ntoa(g.groupAddress())
			   << ", " << g.port() << ", ";
  if (g.isSSM()) {
    return s1 << "SSM source: "
	      <<  our_inet_ntoa(g.sourceFilterAddress()) << ")";
  } else {
    return s1 << (unsigned)(g.ttl()) << ")";
  }
}


////////// GroupsockLookupTable //////////


// A hash table used to index Groupsocks by socket number.

static HashTable* getSocketTable(UsageEnvironment& env) {
  if (env.groupsockPriv == NULL) { // We need to create it
    env.groupsockPriv = HashTable::create(ONE_WORD_HASH_KEYS);
  }
  return (HashTable*)(env.groupsockPriv);
}

static Boolean unsetGroupsockBySocket(Groupsock const* groupsock) {
  do {
    if (groupsock == NULL) break;
    
    int sock = groupsock->socketNum();
    // Make sure "sock" is in bounds:
    if (sock < 0) break;
    
    HashTable* sockets = getSocketTable(groupsock->env());
    if (sockets == NULL) break;
    
    Groupsock* gs = (Groupsock*)sockets->Lookup((char*)(long)sock);
    if (gs == NULL || gs != groupsock) break;
    sockets->Remove((char*)(long)sock);
    
    if (sockets->IsEmpty()) {
      // We can also delete the table (to reclaim space):
      delete sockets;
      (gs->env()).groupsockPriv = NULL;
    }

    return True;
  } while (0);
  
  return False;
}

static Boolean setGroupsockBySocket(UsageEnvironment& env, int sock,
				    Groupsock* groupsock) {
  do {
    // Make sure the "sock" parameter is in bounds:
    if (sock < 0) {
      char buf[100];
      sprintf(buf, "trying to use bad socket (%d)", sock);
      env.setResultMsg(buf);
      break;
    }
    
    HashTable* sockets = getSocketTable(env);
    if (sockets == NULL) break;
    
    // Make sure we're not replacing an existing Groupsock
    // That shouldn't happen
    Boolean alreadyExists
      = (sockets->Lookup((char*)(long)sock) != 0);
    if (alreadyExists) {
      char buf[100];
      sprintf(buf,
	      "Attempting to replace an existing socket (%d",
	      sock);
      env.setResultMsg(buf);
      break;
    }
    
    sockets->Add((char*)(long)sock, groupsock);
    return True;
  } while (0);
  
  return False;
}

static Groupsock* getGroupsockBySocket(UsageEnvironment& env, int sock) {
  do {
    // Make sure the "sock" parameter is in bounds:
    if (sock < 0) break;
    
    HashTable* sockets = getSocketTable(env);
    if (sockets == NULL) break;
    
    return (Groupsock*)sockets->Lookup((char*)(long)sock);
  } while (0);
  
  return NULL;
}

Groupsock*
GroupsockLookupTable::Fetch(UsageEnvironment& env,
			    netAddressBits groupAddress,
			    Port port, u_int8_t ttl,
			    Boolean& isNew) {
  isNew = False;
  Groupsock* groupsock;
  do {
    groupsock = (Groupsock*) fTable.Lookup(groupAddress, (~0), port);
    if (groupsock == NULL) { // we need to create one:
      groupsock = AddNew(env, groupAddress, (~0), port, ttl);
      if (groupsock == NULL) break;
      isNew = True;
    }
  } while (0);
  
  return groupsock;
}

Groupsock*
GroupsockLookupTable::Fetch(UsageEnvironment& env,
			    netAddressBits groupAddress,
			    netAddressBits sourceFilterAddr, Port port,
			    Boolean& isNew) {
  isNew = False;
  Groupsock* groupsock;
  do {
    groupsock
      = (Groupsock*) fTable.Lookup(groupAddress, sourceFilterAddr, port);
    if (groupsock == NULL) { // we need to create one:
      groupsock = AddNew(env, groupAddress, sourceFilterAddr, port, 0);
      if (groupsock == NULL) break;
      isNew = True;
    }
  } while (0);
  
  return groupsock;
}

Groupsock*
GroupsockLookupTable::Lookup(netAddressBits groupAddress, Port port) {
  return (Groupsock*) fTable.Lookup(groupAddress, (~0), port);
}

Groupsock*
GroupsockLookupTable::Lookup(netAddressBits groupAddress,
			     netAddressBits sourceFilterAddr, Port port) {
  return (Groupsock*) fTable.Lookup(groupAddress, sourceFilterAddr, port);
}

Groupsock* GroupsockLookupTable::Lookup(UsageEnvironment& env, int sock) {
  return getGroupsockBySocket(env, sock);
}

Boolean GroupsockLookupTable::Remove(Groupsock const* groupsock) {
  unsetGroupsockBySocket(groupsock);
  return fTable.Remove(groupsock->groupAddress().s_addr,
		       groupsock->sourceFilterAddress().s_addr,
		       groupsock->port());
}

Groupsock* GroupsockLookupTable::AddNew(UsageEnvironment& env,
					netAddressBits groupAddress,
					netAddressBits sourceFilterAddress,
					Port port, u_int8_t ttl) {
  Groupsock* groupsock;
  do {
    struct in_addr groupAddr; groupAddr.s_addr = groupAddress;
    if (sourceFilterAddress == netAddressBits(~0)) {
      // regular, ISM groupsock
      groupsock = new Groupsock(env, groupAddr, port, ttl);
    } else {
      // SSM groupsock
      struct in_addr sourceFilterAddr;
      sourceFilterAddr.s_addr = sourceFilterAddress;
      groupsock = new Groupsock(env, groupAddr, sourceFilterAddr, port);
    }
    
    if (groupsock == NULL || groupsock->socketNum() < 0) break;
    
    if (!setGroupsockBySocket(env, groupsock->socketNum(), groupsock)) break;
    
    fTable.Add(groupAddress, sourceFilterAddress, port, (void*)groupsock);
  } while (0);
  
  return groupsock;
}

GroupsockLookupTable::Iterator::Iterator(GroupsockLookupTable& groupsocks)
  : fIter(AddressPortLookupTable::Iterator(groupsocks.fTable)) {
}

Groupsock* GroupsockLookupTable::Iterator::next() {
  return (Groupsock*) fIter.next();
};