continuous-integration/drone/push Build encountered an error
- src/buffer_state.go replaces the two sync.Maps plus a global RWMutex with one bufferMu guarding a map of *Playlist. Each stream has one shared bufferStream (URL, folder, status, client count, error, cancel hook); downloaders keep a private ThisStream and publish through helpers. Playlist.Clients / ThisClient / ClientConnection are gone: there was one client counter per stream in two places that could disagree. - bufferAcquireStream does the tuner check and the registration under the same lock, so two clients tuning at once cannot both pass the limit. bufferReleaseClient removes the stream when the last client leaves, cancels its process and deletes its segment folder. - bufferingStream rewritten: waits on r.Context() instead of the deprecated CloseNotifier, sets Content-Type before WriteHeader (the old code set headers after and one was literally named "Content-Length:"), flushes each segment to the client, no defer inside the segment loop. - connectToStreamingServer: shared streamHTTPClient with dial, TLS and response-header timeouts (no overall timeout, bodies are endless); the deferred Body/segment closes inside the redirect and read loops are now explicit closes, so a multi-hour stream no longer accumulates them. - thirdPartyBuffer rewritten around exec.CommandContext: the process is killed when the last client leaves or when no usable data arrives within 20 s (time.AfterFunc watchdog, no leaked goroutine); cmd.Start error is checked; no panic; one file handle per segment. ffmpeg and VLC command lines come from buildFFmpegArgs / buildVLCArgs, which are unit tested. - Data.Cache.StreamingURLS is guarded by streamingURLsMu and persisted from a snapshot. - Tests (src/buffer_test.go, run with -race): restream shares one provider connection, six concurrent tunes against a tuner limit of two, 50-way acquire/release contention, cleanup and cancel on last release.
237 lines
5.1 KiB
Go
237 lines
5.1 KiB
Go
package src
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"encoding/xml"
|
|
"fmt"
|
|
)
|
|
|
|
func makeInteraceFromHDHR(content []byte, playlistName, id string) (channels []any, err error) {
|
|
|
|
var hdhrData []any
|
|
|
|
err = json.Unmarshal(content, &hdhrData)
|
|
if err == nil {
|
|
|
|
for _, d := range hdhrData {
|
|
|
|
var channel = make(map[string]string)
|
|
var data = d.(map[string]any)
|
|
|
|
channel["group-title"] = playlistName
|
|
channel["name"] = data["GuideName"].(string)
|
|
channel["tvg-id"] = data["GuideName"].(string)
|
|
channel["url"] = data["URL"].(string)
|
|
channel["ID-"+id] = data["GuideNumber"].(string)
|
|
channel["_uuid.key"] = "ID-" + id
|
|
channel["_values"] = playlistName + " " + channel["name"]
|
|
|
|
channels = append(channels, channel)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func getCapability() (xmlContent []byte, err error) {
|
|
|
|
var capability Capability
|
|
var buffer bytes.Buffer
|
|
|
|
capability.Xmlns = "urn:schemas-upnp-org:device-1-0"
|
|
capability.URLBase = System.ServerProtocol.WEB + "://" + System.Domain
|
|
|
|
capability.SpecVersion.Major = 1
|
|
capability.SpecVersion.Minor = 0
|
|
|
|
capability.Device.DeviceType = "urn:schemas-upnp-org:device:MediaServer:1"
|
|
capability.Device.FriendlyName = System.Name
|
|
capability.Device.Manufacturer = "Silicondust"
|
|
capability.Device.ModelName = "HDTC-2US"
|
|
capability.Device.ModelNumber = "HDTC-2US"
|
|
capability.Device.SerialNumber = ""
|
|
capability.Device.UDN = "uuid:" + System.DeviceID
|
|
|
|
output, err := xml.MarshalIndent(capability, " ", " ")
|
|
if err != nil {
|
|
ShowError(err, 1003)
|
|
}
|
|
|
|
buffer.Write([]byte(xml.Header))
|
|
buffer.Write([]byte(output))
|
|
xmlContent = buffer.Bytes()
|
|
|
|
return
|
|
}
|
|
|
|
func getDiscover() (jsonContent []byte, err error) {
|
|
|
|
var discover Discover
|
|
|
|
discover.BaseURL = System.ServerProtocol.WEB + "://" + System.Domain
|
|
discover.DeviceAuth = System.AppName
|
|
discover.DeviceID = System.DeviceID
|
|
discover.FirmwareName = "bin_" + System.Version
|
|
discover.FirmwareVersion = System.Version
|
|
discover.FriendlyName = System.Name
|
|
|
|
discover.LineupURL = fmt.Sprintf("%s://%s/lineup.json", System.ServerProtocol.DVR, System.Domain)
|
|
discover.Manufacturer = "Golang"
|
|
discover.ModelNumber = System.Version
|
|
discover.TunerCount = Settings.Tuner
|
|
|
|
jsonContent, err = json.MarshalIndent(discover, "", " ")
|
|
|
|
return
|
|
}
|
|
|
|
func getLineupStatus() (jsonContent []byte, err error) {
|
|
|
|
var lineupStatus LineupStatus
|
|
|
|
lineupStatus.ScanInProgress = System.ScanInProgress
|
|
lineupStatus.ScanPossible = 0
|
|
lineupStatus.Source = "Cable"
|
|
lineupStatus.SourceList = []string{"Cable"}
|
|
|
|
jsonContent, err = json.MarshalIndent(lineupStatus, "", " ")
|
|
|
|
return
|
|
}
|
|
|
|
func getLineup() (jsonContent []byte, err error) {
|
|
|
|
var lineup Lineup
|
|
|
|
switch Settings.EpgSource {
|
|
|
|
case "PMS":
|
|
for i, dsa := range Data.Streams.Active {
|
|
|
|
var m3uChannel M3UChannelStructXEPG
|
|
|
|
err = json.Unmarshal([]byte(mapToJSON(dsa)), &m3uChannel)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
var stream LineupStream
|
|
stream.GuideName = m3uChannel.Name
|
|
switch len(m3uChannel.UUIDValue) {
|
|
|
|
case 0:
|
|
stream.GuideNumber = fmt.Sprintf("%d", i+1000)
|
|
guideNumber, err := getGuideNumberPMS(stream.GuideName)
|
|
if err != nil {
|
|
ShowError(err, 0)
|
|
}
|
|
|
|
stream.GuideNumber = guideNumber
|
|
|
|
default:
|
|
stream.GuideNumber = m3uChannel.UUIDValue
|
|
|
|
}
|
|
|
|
stream.URL, err = createStreamingURL("DVR", m3uChannel.FileM3UID, stream.GuideNumber, m3uChannel.Name, m3uChannel.URL)
|
|
if err == nil {
|
|
lineup = append(lineup, stream)
|
|
} else {
|
|
ShowError(err, 1202)
|
|
}
|
|
|
|
}
|
|
|
|
case "XEPG":
|
|
for _, dxc := range Data.XEPG.Channels {
|
|
|
|
var xepgChannel XEPGChannelStruct
|
|
err = json.Unmarshal([]byte(mapToJSON(dxc)), &xepgChannel)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
if xepgChannel.XActive {
|
|
var stream LineupStream
|
|
stream.GuideName = xepgChannel.XName
|
|
stream.GuideNumber = xepgChannel.XChannelID
|
|
//stream.URL = fmt.Sprintf("%s://%s/stream/%s-%s", System.ServerProtocol.DVR, System.Domain, xepgChannel.FileM3UID, base64.StdEncoding.EncodeToString([]byte(xepgChannel.URL)))
|
|
stream.URL, err = createStreamingURL("DVR", xepgChannel.FileM3UID, xepgChannel.XChannelID, xepgChannel.XName, xepgChannel.URL)
|
|
if err == nil {
|
|
lineup = append(lineup, stream)
|
|
} else {
|
|
ShowError(err, 1202)
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
jsonContent, err = json.MarshalIndent(lineup, "", " ")
|
|
|
|
Data.Cache.PMS = nil
|
|
|
|
saveMapToJSONFile(System.File.URLS, streamingURLsSnapshot())
|
|
|
|
return
|
|
}
|
|
|
|
func getGuideNumberPMS(channelName string) (pmsID string, err error) {
|
|
|
|
if len(Data.Cache.PMS) == 0 {
|
|
|
|
Data.Cache.PMS = make(map[string]string)
|
|
|
|
pms, err := loadJSONFileToMap(System.File.PMS)
|
|
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
for key, value := range pms {
|
|
Data.Cache.PMS[key] = value.(string)
|
|
}
|
|
|
|
}
|
|
|
|
var getNewID = func(channelName string) (id string) {
|
|
|
|
var i int
|
|
|
|
newID:
|
|
|
|
var ids []string
|
|
id = fmt.Sprintf("id-%d", i)
|
|
|
|
for _, v := range Data.Cache.PMS {
|
|
ids = append(ids, v)
|
|
}
|
|
|
|
if indexOfString(id, ids) != -1 {
|
|
i++
|
|
goto newID
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
if value, ok := Data.Cache.PMS[channelName]; ok {
|
|
|
|
pmsID = value
|
|
|
|
} else {
|
|
|
|
pmsID = getNewID(channelName)
|
|
Data.Cache.PMS[channelName] = pmsID
|
|
saveMapToJSONFile(System.File.PMS, Data.Cache.PMS)
|
|
|
|
}
|
|
|
|
return
|
|
}
|