Add mutex for traffic/network ops.

network.go

@@ -6,12 +6,14 @@ import (
 	"net"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
 var messageQueue chan []byte
 var outSockets map[string]*net.UDPConn
 var dhcpLeases map[string]string
+var netMutex *sync.Mutex
 
 // Read the "dhcpd.leases" file and parse out IP/hostname.
 func getDHCPLeases() (map[string]string, error) {
@@ -38,6 +40,8 @@ func getDHCPLeases() (map[string]string, error) {
 }
 
 func sendToAllConnectedClients(msg []byte) {
+	netMutex.Lock()
+	defer netMutex.Unlock()
 	for _, sock := range outSockets {
 		sock.Write(msg)
 	}
@@ -50,6 +54,8 @@ func getNetworkStats() int {
 
 // See who has a DHCP lease and make a UDP connection to each of them.
 func refreshConnectedClients() {
+	netMutex.Lock()
+	defer netMutex.Unlock()
 	validConnections := make(map[string]bool)
 	t, err := getDHCPLeases()
 	if err != nil {
@@ -111,6 +117,7 @@ func sendMsg(msg []byte) {
 func initNetwork() {
 	messageQueue = make(chan []byte, 1024) // Buffered channel, 1024 messages.
 	outSockets = make(map[string]*net.UDPConn)
+	netMutex = &sync.Mutex{}
 	refreshConnectedClients()
 	go messageQueueSender()
 }

traffic.go

@@ -3,12 +3,12 @@ package main
 import (
 	"encoding/hex"
 	"math"
+	"sync"
 	"time"
 )
 
 //-0b2b48fe3aef1f88621a0856110a31c01105c4e6c4e6c40a9a820300000000000000;rs=7;
 
-
 /*
 
 HDR:
@@ -36,8 +36,8 @@ MS:
  NACp:              8
  NACv:              1
  NICbaro:           0
- Capabilities:      
+ Capabilities:
- Active modes:      
+ Active modes:
  Target track type: true heading
 AUXSV:
  Sec. altitude:     unavailable
@@ -45,30 +45,33 @@ AUXSV:
 */
 
 type TrafficInfo struct {
-	icao_addr		uint32
+	icao_addr uint32
-	addr_type 		uint8
+	addr_type uint8
 
-	lat				float32
+	lat float32
-	lng				float32
+	lng float32
 
-	position_valid	bool
+	position_valid bool
 
-	alt				uint32
+	alt uint32
 
-	track			uint16
+	track       uint16
-	speed			uint16
+	speed       uint16
-	speed_valid		bool
+	speed_valid bool
 
-	vvel			int16
+	vvel int16
 
-	tail			string
+	tail string
 
-	last_seen		time.Time
+	last_seen time.Time
 }
 
 var traffic map[uint32]TrafficInfo
+var trafficMutex *sync.Mutex
 
 func cleanupOldEntries() {
+	trafficMutex.Lock()
+	defer trafficMutex.Unlock()
 	for icao_addr, ti := range traffic {
 		if time.Since(ti.last_seen).Seconds() > float64(60) { //FIXME: 60 seconds with no update on this address - stop displaying.
 			delete(traffic, icao_addr)
@@ -77,6 +80,8 @@ func cleanupOldEntries() {
 }
 
 func sendTrafficUpdates() {
+	trafficMutex.Lock()
+	defer trafficMutex.Unlock()
 	cleanupOldEntries()
 	for _, ti := range traffic {
 		makeTrafficReport(ti)
@@ -85,6 +90,7 @@ func sendTrafficUpdates() {
 
 func initTraffic() {
 	traffic = make(map[uint32]TrafficInfo)
+	trafficMutex = &sync.Mutex{}
 }
 
 func makeTrafficReport(ti TrafficInfo) {
@@ -109,15 +115,14 @@ func makeTrafficReport(ti TrafficInfo) {
 	lng := float32(ti.lng)
 	tmp = makeLatLng(lng)
 
-	msg[8] = tmp[0] // Longitude.
+	msg[8] = tmp[0]  // Longitude.
-	msg[9] = tmp[1] // Longitude.
+	msg[9] = tmp[1]  // Longitude.
 	msg[10] = tmp[2] // Longitude.
 
-
+	//Altitude: OK
-//Altitude: OK
+	//TODO: 0xFFF "invalid altitude."
-//TODO: 0xFFF "invalid altitude."
 	alt := uint16(ti.alt)
-	alt = (alt + 1000)/25
+	alt = (alt + 1000) / 25
 	alt = alt & 0xFFF // Should fit in 12 bits.
 
 	msg[11] = byte((alt & 0xFF0) >> 4) // Altitude.
@@ -130,11 +135,11 @@ func makeTrafficReport(ti TrafficInfo) {
 	// Horizontal velocity (speed).
 
 	msg[14] = byte((ti.speed & 0x0FF0) >> 4)
-	msg[15] = byte((ti.speed & 0x000F) << 4) 
+	msg[15] = byte((ti.speed & 0x000F) << 4)
 
 	// Vertical velocity.
 	vvel := ti.vvel / 64 // 64fpm resolution.
-	msg[15] = msg[15] | byte((vvel & 0x0F00) >> 8)
+	msg[15] = msg[15] | byte((vvel&0x0F00)>>8)
 	msg[16] = byte(vvel & 0x00FF)
 
 	// Track.
@@ -146,7 +151,6 @@ func makeTrafficReport(ti TrafficInfo) {
 	sendMsg(prepareMessage(msg))
 }
 
-
 func parseDownlinkReport(s string) {
 	var ti TrafficInfo
 	s = s[1:]
@@ -154,12 +158,12 @@ func parseDownlinkReport(s string) {
 	hex.Decode(frame, []byte(s))
 
 	// Header.
-//	msg_type := (uint8(frame[0]) >> 3) & 0x1f
+	//	msg_type := (uint8(frame[0]) >> 3) & 0x1f
 	ti.addr_type = uint8(frame[0]) & 0x07
 	ti.icao_addr = (uint32(frame[1]) << 16) | (uint32(frame[2]) << 8) | uint32(frame[3])
 
 	// OK.
-//	fmt.Printf("%d, %d, %06X\n", msg_type, ti.addr_type, ti.icao_addr)
+	//	fmt.Printf("%d, %d, %06X\n", msg_type, ti.addr_type, ti.icao_addr)
 
 	nic := uint8(frame[11]) & 15 //TODO: Meaning?
 
@@ -186,20 +190,20 @@ func parseDownlinkReport(s string) {
 	ti.position_valid = position_valid
 
 	raw_alt := (uint32(frame[10]) << 4) | ((uint32(frame[11]) & 0xf0) >> 4)
-//	alt_geo := false // Barometric if not geometric.
+	//	alt_geo := false // Barometric if not geometric.
 	alt := uint32(0)
 	if raw_alt != 0 {
-//		alt_geo = (uint8(frame[9]) & 1) != 0
+		//		alt_geo = (uint8(frame[9]) & 1) != 0
 		alt = ((raw_alt - 1) * 25) - 1000
 	}
 	ti.alt = alt
 
 	//OK.
-//	fmt.Printf("%d, %t, %f, %f, %t, %d\n", nic, position_valid, lat, lng, alt_geo, alt)
+	//	fmt.Printf("%d, %t, %f, %f, %t, %d\n", nic, position_valid, lat, lng, alt_geo, alt)
 
 	airground_state := (uint8(frame[12]) >> 6) & 0x03
 	//OK.
-//	fmt.Printf("%d\n", airground_state)
+	//	fmt.Printf("%d\n", airground_state)
 
 	ns_vel := int16(0)
 	ew_vel := int16(0)
@@ -207,7 +211,7 @@ func parseDownlinkReport(s string) {
 	speed_valid := false
 	speed := uint16(0)
 	vvel := int16(0)
-//	vvel_geo := false
+	//	vvel_geo := false
 	if airground_state == 0 || airground_state == 1 { // Subsonic. Supersonic.
 		// N/S velocity.
 		ns_vel_valid := false
@@ -215,14 +219,14 @@ func parseDownlinkReport(s string) {
 		raw_ns := ((int16(frame[12]) & 0x1f) << 6) | ((int16(frame[13]) & 0xfc) >> 2)
 		if (raw_ns & 0x3ff) != 0 {
 			ns_vel_valid = true
-            ns_vel = ((raw_ns & 0x3ff) - 1)
+			ns_vel = ((raw_ns & 0x3ff) - 1)
-            if (raw_ns & 0x400) != 0 {
+			if (raw_ns & 0x400) != 0 {
-                ns_vel = 0 - ns_vel
+				ns_vel = 0 - ns_vel
 			}
-            if airground_state == 1 { // Supersonic.
+			if airground_state == 1 { // Supersonic.
-                ns_vel = ns_vel * 4
+				ns_vel = ns_vel * 4
 			}
-        }
+		}
 		// E/W velocity.
 		raw_ew := ((int16(frame[13]) & 0x03) << 9) | (int16(frame[14]) << 1) | ((int16(frame[15] & 0x80)) >> 7)
 		if (raw_ew & 0x3ff) != 0 {
@@ -238,7 +242,7 @@ func parseDownlinkReport(s string) {
 		if ns_vel_valid && ew_vel_valid {
 			if ns_vel != 0 && ew_vel != 0 {
 				//TODO: Track type
-				track = (360 + 90 - uint16(math.Atan2(float64(ns_vel), float64(ew_vel)) * 180 / math.Pi)) % 360
+				track = (360 + 90 - uint16(math.Atan2(float64(ns_vel), float64(ew_vel))*180/math.Pi)) % 360
 			}
 			speed_valid = true
 			speed = uint16(math.Sqrt(float64((ns_vel * ns_vel) + (ew_vel * ew_vel))))
@@ -247,7 +251,7 @@ func parseDownlinkReport(s string) {
 		// Vertical velocity.
 		raw_vvel := ((int16(frame[15]) & 0x7f) << 4) | ((int16(frame[16]) & 0xf0) >> 4)
 		if (raw_vvel & 0x1ff) != 0 {
-//			vvel_geo = (raw_vvel & 0x400) == 0
+			//			vvel_geo = (raw_vvel & 0x400) == 0
 			vvel = ((raw_vvel & 0x1ff) - 1) * 64
 			if (raw_vvel & 0x200) != 0 {
 				vvel = 0 - vvel
@@ -264,23 +268,93 @@ func parseDownlinkReport(s string) {
 	ti.speed_valid = speed_valid
 
 	//OK.
-//	fmt.Printf("ns_vel %d, ew_vel %d, track %d, speed_valid %t, speed %d, vvel_geo %t, vvel %d\n", ns_vel, ew_vel, track, speed_valid, speed, vvel_geo, vvel)
+	//	fmt.Printf("ns_vel %d, ew_vel %d, track %d, speed_valid %t, speed %d, vvel_geo %t, vvel %d\n", ns_vel, ew_vel, track, speed_valid, speed, vvel_geo, vvel)
 
-/*
+	/*
-	utc_coupled := false
+		utc_coupled := false
-	tisb_site_id := uint8(0)
+		tisb_site_id := uint8(0)
 
-	if (uint8(frame[0]) & 7) == 2 || (uint8(frame[0]) & 7) == 3 { //TODO: Meaning?
+		if (uint8(frame[0]) & 7) == 2 || (uint8(frame[0]) & 7) == 3 { //TODO: Meaning?
-		tisb_site_id = uint8(frame[16]) & 0x0f
+			tisb_site_id = uint8(frame[16]) & 0x0f
-	} else {
+		} else {
-		utc_coupled = (uint8(frame[16]) & 0x08) != 0
+			utc_coupled = (uint8(frame[16]) & 0x08) != 0
-	}
+		}
-*/
+	*/
 
 	//OK.
-//	fmt.Printf("tisb_site_id %d, utc_coupled %t\n", tisb_site_id, utc_coupled)
+	//	fmt.Printf("tisb_site_id %d, utc_coupled %t\n", tisb_site_id, utc_coupled)
 
 	ti.last_seen = time.Now()
 
+	trafficMutex.Lock()
 	traffic[ti.icao_addr] = ti
-}
+	trafficMutex.Unlock()
+}
+
+//TODO
+/*
+//	dump1090Addr    = "127.0.0.1:30003"
+inConn, err := net.Dial("tcp", dump1090Addr)
+if err != nil {
+	time.Sleep(1 * time.Second)
+	continue
+}
+rdr := bufio.NewReader(inConn)
+for {
+	buf, err := rdr.ReadString('\n')
+	if err != nil { // Must have disconnected?
+		break
+	}
+	buf = strings.Trim(buf, "\r\n")
+	//log.Printf("%s\n", buf)
+	x := strings.Split(buf, ",")
+	//TODO: Add more sophisticated stuff that combines heading/speed updates with the location.
+	if len(x) < 22 {
+		continue
+	}
+	icao := x[4]
+	icaoDec, err := strconv.ParseInt(icao, 16, 32)
+	if err != nil {
+		continue
+	}
+	mutex.Lock()
+	// Retrieve previous information on this ICAO code.
+	var pi PositionInfo
+	if _, ok := blips[icaoDec]; ok {
+		pi = blips[icaoDec]
+	}
+
+	if x[1] == "3" {
+		//MSG,3,111,11111,AC2BB7,111111,2015/07/28,03:59:12.363,2015/07/28,03:59:12.353,,5550,,,42.35847,-83.42212,,,,,,0
+		alt := x[11]
+		lat := x[14]
+		lng := x[15]
+
+		//log.Printf("icao=%s, icaoDec=%d, alt=%s, lat=%s, lng=%s\n", icao, icaoDec, alt, lat, lng)
+		pi.alt = alt
+		pi.lat = lat
+		pi.lng = lng
+	}
+	if x[1] == "4" {
+		// MSG,4,111,11111,A3B557,111111,2015/07/28,06:13:36.417,2015/07/28,06:13:36.398,,,414,278,,,-64,,,,,0
+		vel := x[12]
+		hdg := x[13]
+		vr := x[16]
+
+		//log.Printf("icao=%s, icaoDec=%d, vel=%s, hdg=%s, vr=%s\n", icao, icaoDec, vel, hdg, vr)
+		pi.vel = vel
+		pi.hdg = hdg
+		pi.vr = vr
+	}
+	if x[1] == "1" {
+		// MSG,1,,,%02X%02X%02X,,,,,,%s,,,,,,,,0,0,0,0
+		tail := x[10]
+		pi.tail = tail
+	}
+
+	// Update "last seen" (any type of message).
+	pi.last_seen = time.Now()
+
+	blips[icaoDec] = pi // Update information on this ICAO code.
+	mutex.Unlock()
+*/