Address races for conn.connected.

[stompngo.git] / connection.go

//
// Copyright © 2011-2019 Guy M. Allard
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//

package stompngo

import (
        "log"
        "runtime"
        "time"
)

// Exported Connection methods

/*
        Connected returns the current connection status.
*/
func (c *Connection) Connected() bool {
        return c.isConnected()
}

/*
        Session returns the broker assigned session id.
*/
func (c *Connection) Session() string {
        return c.session
}

/*
        Protocol returns the current connection protocol level.
*/
func (c *Connection) Protocol() string {
        c.protoLock.Lock()
        defer c.protoLock.Unlock()
        return c.protocol
}

/*
        SetLogger enables a client defined logger for this connection.

        Set to "nil" to disable logging.

        Example:
                // Start logging
                l := log.New(os.Stdout, "", log.Ldate|log.Lmicroseconds)
                c.SetLogger(l)
*/
func (c *Connection) SetLogger(l *log.Logger) {
        logLock.Lock()
        c.logger = l
        logLock.Unlock()
}

/*
        GetLogger - returns the current connection logger.
*/
func (c *Connection) GetLogger() *log.Logger {
        return c.logger
}

/*
        SendTickerInterval returns any heartbeat send ticker interval in ms.  A return
        value of zero means     no heartbeats are being sent.
*/
func (c *Connection) SendTickerInterval() int64 {
        if c.hbd == nil {
                return 0
        }
        return c.hbd.sti / 1000000
}

/*
        ReceiveTickerInterval returns any heartbeat receive ticker interval in ms.
        A return value of zero means no heartbeats are being received.
*/
func (c *Connection) ReceiveTickerInterval() int64 {
        if c.hbd == nil {
                return 0
        }
        return c.hbd.rti / 1000000
}

/*
        SendTickerCount returns any heartbeat send ticker count.  A return value of
        zero usually indicates no send heartbeats are enabled.
*/
func (c *Connection) SendTickerCount() int64 {
        if c.hbd == nil {
                return 0
        }
        return c.hbd.sc
}

/*
        ReceiveTickerCount returns any heartbeat receive ticker count. A return
        value of zero usually indicates no read heartbeats are enabled.
*/
func (c *Connection) ReceiveTickerCount() int64 {
        if c.hbd == nil {
                return 0
        }
        return c.hbd.rc
}

/*
        FramesRead returns a count of the number of frames read on the connection.
*/
func (c *Connection) FramesRead() int64 {
        return c.mets.tfr
}

/*
        BytesRead returns a count of the number of bytes read on the connection.
*/
func (c *Connection) BytesRead() int64 {
        return c.mets.tbr
}

/*
        FramesWritten returns a count of the number of frames written on the connection.
*/
func (c *Connection) FramesWritten() int64 {
        return c.mets.tfw
}

/*
        BytesWritten returns a count of the number of bytes written on the connection.
*/
func (c *Connection) BytesWritten() int64 {
        return c.mets.tbw
}

/*
        Running returns a time duration since connection start.
*/
func (c *Connection) Running() time.Duration {
        return time.Since(c.mets.st)
}

/*
        SubChanCap returns the current subscribe channel capacity.
*/
func (c *Connection) SubChanCap() int {
        return c.scc
}

/*
        SetSubChanCap sets a new subscribe channel capacity, to be used during future
        SUBSCRIBE operations.
*/
func (c *Connection) SetSubChanCap(nc int) {
        c.scc = nc
        return
}

// Unexported Connection methods

/*
        Log data if possible.
*/
func (c *Connection) log(v ...interface{}) {
        logLock.Lock()
        defer logLock.Unlock()
        if c.logger == nil {
                return
        }
        _, fn, ld, ok := runtime.Caller(1)

        if ok {
                c.logger.Printf("%s %s %d %v\n", c.session, fn, ld, v)
        } else {
                c.logger.Printf("%s %v\n", c.session, v)
        }
        return
}

/*
        Log data if possible (extended / abbreviated) logic).
*/
func (c *Connection) logx(v ...interface{}) {
        _, fn, ld, ok := runtime.Caller(1)

        c.sessLock.Lock()
        if ok {
                c.logger.Printf("%s %s %d %v\n", c.session, fn, ld, v)
        } else {
                c.logger.Printf("%s %v\n", c.session, v)
        }
        c.sessLock.Unlock()
        return
}

/*
        Shutdown heartbeats
*/
func (c *Connection) shutdownHeartBeats() {
        // Shutdown heartbeats if necessary
        if c.hbd != nil {
                c.hbd.clk.Lock()
                if !c.hbd.ssdn {
                        if c.hbd.hbs {
                                close(c.hbd.ssd)
                        }
                        if c.hbd.hbr {
                                close(c.hbd.rsd)
                        }
                        c.hbd.ssdn = true
                }
                c.hbd.clk.Unlock()
        }
}

/*
        Shutdown logic.
*/
func (c *Connection) shutdown() {
        c.log("SHUTDOWN", "starts")
        c.shutdownHeartBeats()
        // Close all individual subscribe channels
        // This is a write lock
        c.subsLock.Lock()
        for key := range c.subs {
                close(c.subs[key].md)
                c.subs[key].cs = true
        }
        c.setConnected(false)
        c.subsLock.Unlock()
        c.log("SHUTDOWN", "ends")
        return
}

/*
        Connection Abort logic. Shutdown connection system on problem happens
*/
func (c *Connection) sysAbort() {
        c.abortOnce.Do(func() { close(c.ssdc) })
        return
}

/*
        Read error handler.
*/
func (c *Connection) handleReadError(md MessageData) {
        c.log("HDRERR", "starts", md)
        c.shutdownHeartBeats() // We are done here
        // Notify any general subscriber of error
        select {
        case c.input <- md:
        default:
        }
        // Notify all individual subscribers of error
        // This is a read lock
        c.subsLock.RLock()
        if c.isConnected() {
                for key := range c.subs {
                        c.subs[key].md <- md
                }
        }
        c.subsLock.RUnlock()
        // Try to catch the writer
        close(c.wtrsdc)
        c.log("HDRERR", "ends")
        // Let further shutdown logic proceed normally.
        return
}

/*
        Connected check
*/
func (c *Connection) isConnected() bool {
        c.connLock.Lock()
        defer c.connLock.Unlock()
        return c.connected
}

/*
        Connected set
*/
func (c *Connection) setConnected(to bool) {
        c.connLock.Lock()
        defer c.connLock.Unlock()
        c.connected = to
}