sendmail V2

/**
* 
* https://github.com/lgkahn/hubitat/blob/master/sendmail%20V2
* 
* v 2.1 added desciprtive text and cleaned up debugging
* v 2.2 added hostname option and lookup ip if blank.
* v 2.3 added better checking for end of send to make sure it is closing cleanly
* v 2.4 add serial queueing so that a new connect waits while old finishes.. set timeout to 1 minute to wait. if email is not finished by
*       then close connection and start new one. Also add state.lastCommand to follow the attribute as that seems more reliable.
* v 2.4.1 ignore stream is closed errors in telnetStatus fx as the is a system callback and is not working correctly , intermittantly i get a call
*       to this function telling me socket is closed but it obviously is not because responses are still asynchronously coming basck after!
*      This "Stream is closed" messages seems to always come out after the quit command so it appears to be a status message not really an error
* v 2.4.2. try to workaround the check i had for the goodbye/bye message .. I was checkiong the value string on code 221 which was not a good
*          approach as various servers can return differnent strings here. Unfortunately 221 just means remote server is closing so need to checkt
*          that the message was sent and it is 221,
* v 2.4.3 true concurrency and queue using a thread safe data structure concurrentlinkedqueue and also mutex (thanks to erktrek)
*         tested with up to 8 messages
*         When one message finishes it checks the queue, and delivers any other remaining.. also schedules a rerun when busy and one is added to the queue.
*         also serialize the setting and checking of the state.lastCommand as this is used to keep track of the async status of the commands
* 
*  v 2.5 changes were enough to signify new version. Found one bug in last one when reviewing code. a synchronize inside a synchronize on same semaphore.
*         Did not seem to be causing issues so I assume the system is smart enough to avoid it but fixed anyway.
* v 2.5.1 mine got stuck in weird state and kept re-running with failure.. added an unschedule to fix it in certain cases and reset states. Also removed a function no longer called.
*         Also reset state variable when it finds queue empty.
* v 2.5.2 change password input type to password from text
* v 2.5.3 auto turn off logging after 1/2 hour
* v 2.5.4 change formatting of date header
* v 2.5.5 change time formatting yet again.
* v 2.5.6 put extra lines in send to try to get around error in hubitat 2.2.5
* v 2.5.7 removed extra lines as bug is 2.2.5 and nothing to do with the lines.
* v 2.5.8 fixed email.. as summized version 2.2.5 telnet is stripping of extra lf or cr that are needed for email
* v 2.6 played with carriage return line feeds to get message back on first line..
* v 3.0 after recent release hubitat seems much slower getting the initial connect to the telnet port and therefore messages are stacking up .
* for this reason instead of queuing and trying 30 seconds later I have added a random component so it will queue and try 30 + 1-60 secs later.
* v 3.1 add desc. logging option default is on.. only a few lines of info come out when debugging is off. One for each email and a couple of others if queued up.
* this turns all that off.
* NOte: just noticed that if you send a bunch of message to the queue and also at the same time to another instance of the driver (ie another notification device) the queue
* is not distince. Meaining that all run in one process/thread.. The assumption is that each would be independent. Not the case . So the wrong device can get a message sent but it is rare.
* will look at somehow appending the device id to the queue and ignoring those not for you, but that is a big change .
* v 3.2 only one of the /r/n was needed before the body.. m
* v 3.3 got mms working from AT&T yeah.. so that we now can get all the msgs in one thread instead of the random number.
*
* V 3.4
* very very complicated. Spend about 6 hours on it.. here are my findings:
* 1. the date header must be there and in a specific format. It is so picky time zone must be in -0500 etc format not EST.
* 2. all FROM and TO emails must be in the form kahn@lgk.com <kahn@lgk.com>. It automatically constructs this so don't enter them that way.
* 3. there must be headers for Message-ID , also enclosed in <>, MIME-Version, Content-Type, and Content-Transfer-Encoding.
* If any of these are missing or wrong the mms doesn't work from AT&T and you get an error email back.
* This thing is so damn picky and there is no documentaton on it whatsoever.
* Send some Beers my way.. What a pain the Ass Hopefully in works for everyone on your existing servers. If not let me know!
* It was also complicated to generate the unique message id. GetHubUuid() did not work. 
* It works without a unique id, but I tried to generate one anyway. It uses a random UUID from a javascript function
* concatenated with the number of secs since midnight 1972 etc.
*
* v 3.5 Some email servers did not like the "email" <email> format, so trying just <email> in the recpt to and from headers.
*
* v 3.5.1 as someone pointed out the character map on hubitat is utf-8 not ascii so some special characters in emails were not being displayed correctly.
* v 3.6 noticed while i was having internet problems that the app kept spawning off retrying forever.. Notice the 2 attempts was not working.
* fixed it so it gives up after two failed attemps. also unchedules all pending and resets and tries again.
* v 4. add optional ehlo for servers that require it. Also need to handle the additional processing of EHLO informational messages coming back.
* v 4.01 bug found with two typos with misspelling of lastCommand. Fix brought to light corresponding missing 250 checks.
* v 4.02 Replaced descLog with txtEnable and debug with logEnable to conform with built-in drivers and consolidate Hubitat Preference Manager entries for better user experience (2022-10-06)
*/

attribute "lastCommand", "string"
attribute "myHostName", "string"

import java.util.concurrent.Semaphore
import java.util.concurrent.TimeUnit
import groovy.transform.Field
import java.util.concurrent.ConcurrentLinkedQueue

@Field static java.util.concurrent.Semaphore mutex = new java.util.concurrent.Semaphore(1)
@Field static java.util.concurrent.Semaphore lastStateMutex = new java.util.concurrent.Semaphore(1)

@Field static lqueue = new java.util.concurrent.ConcurrentLinkedQueue()

preferences {
	input("EmailServer", "text", title: "Email Server:", description: "Enter location of email server", required: true)
	input("EmailPort", "integer", title: "Port #:", description: "Enter port number, default 25", defaultValue: 25)
	input("From", "text", title: "From:", description: "", required: true)
 	input("To", "text", title: "To:", description: "", required: true)
	input("Subject", "text", title: "Subject:", description: "")
    input("myHostName", "text", title: "Your host name:", description: "Fully qualified domain/hostname (FQDN) to use in initial HELO/EHELO command. If blank you ip address will be used?", required: false)
    input("logEnable", "bool", title: "Enable debug logging", required: false, defaultValue: false)
    input("txtEnable", "bool", title: "Enable descriptionText logging", required: false, defaultValue: true)
    input("Authenticate", "bool", title: "Use Authentication on the server?", required: false, defaultValue: false)
    input("RequiresEHLO", "bool", title: "Does the server require the EHLO command instead of the std HELO?", required: false, defaultValue: false)
    input("Username", "text", title: "Username for Authentication - (base64 encoded)?", required: false, defaultValue: "")
    input("Password", "password", title: "Password for Authentication - (base64 encoded)?", required: false, defaultValue: "")
}

metadata {
    definition (name: "SMS", namespace: "lgkapps", author: "larry kahn kahn@lgk.com") {
    capability "Notification"
    capability "Actuator"
	capability "Telnet"
	capability "Configuration"
    }
}

def installed() {
    initialize()
}

def updated() {
    initialize()   
}

def configure()
{
    initialize()
}

def logsOff()
{
    log.debug "Turning off Logging!"
    device.updateSetting("logEnable",[value:"false",type:"bool"])
}

def initialize() {

    state.lastMsg = ""
	state.LastCode = 0
	state.EmailBody = ""
    state.lastCommand = "quit"
     
    mutex.release()
    
    if ((!myHostName) || (myHostName == ""))
    {
        log.info "User specified hostname is blank using IP instead."
        def hub = location.hubs[0]
	    def myName =  "[" + hub.getDataValue("localIP") + "]"
    
        state.myHostName = myName
        sendEvent(name: "myHostName", value: myName)
    }
    
    else
    {
        state.myHostName = myHostName
        sendEvent(name: "myHostName", value: myHostName) 
    }
    
   if (txtEnable) log.info "Descriptive Text logging is on."
   else log.info "Description Text logging if off."
    
   if (logEnable)
    {
        log.info "Debug logging is on. Turning off debug logging in 1/2 hour."
        runIn(1800,logsOff)
    }
   else log.info "Debug logging is off."
   
}

def restartFromRunIn()
{  
    processQueue(true)
}

def processQueue(Boolean fromRunIn = false)
{
    def Integer waitTime = 30000
    def Boolean doprocess  = false
    def String msg = []
    
     if (logEnable) log.debug "in process queue queue = $lqueue, fromRunIn = $fromRunIn"
      if (mutex.tryAcquire(waitTime,TimeUnit.MILLISECONDS))
       {  
        def isempty = lqueue.isEmpty()
        if (logEnable) log.debug "in process queue current empty = $isempty"

        if (!isempty)
         {
          if (logEnable) 
             {
                 log.debug "Getting item to process"
     	         log.debug "Got mutex"
             }
             
           msg = lqueue.poll()
           if (logEnable) log.debug "Got item $msg to process."
           mutex.release()   
           doProcess = true
         }
       else
           {
               // queue is empty so reset state just in case
               synchronized (lastStateMutex) { state.lastCommand = "Sent Ok" }
               sendEvent(name: "lastCommand", value: "Sent Ok")  
           }
       }        
   
    else
    {
        log.debug "Lock Acquire failed ... Aborting!"
        mutex.release()
        unschedule()
        exit
    }
    
    mutex.release()  
    if (doProcess) deviceNotification(msg,fromRunIn) 
    
    if (logEnable) log.debug "after run queue = $lqueue"
}     

def addToQueue(String message)

{
    if (logEnable) log.debug "in add to queue current queue = $lqueue"
    if (logEnable) log.debug("Acquiring semaphore.")
   
    def Integer  waitTime = 30000
    
    if (mutex.tryAcquire(waitTime,TimeUnit.MILLISECONDS))
    {    
        
       def isempty = lqueue.isEmpty()
       if (logEnable) log.debug "in process queue current empty = $empty"
       lqueue.add(message)
    }
    else
    {
        log.debug "Lock Acquire failed ... Aborting"
        mutex.release()
        unschedule()
        exit
    }
    
    mutex.release()
    if (logEnable) log.debug "after queue = $lqueue"
}

def deviceNotification(String message, Boolean fromRunIn = false) {

def version = "4.01"
def Boolean goOn = true
state.messageSent = false  
sendEvent(name: "telnet", value: "Ok")
  
synchronized (lastStateMutex)
    {
     if (state.lastCommand == null) state.lastCommand = "quit"
    }
    
    if (logEnable) log.debug "-------> In lgk sendmail Version ($version)"
    
    // lgk now check if we are already in middle of a message and if so wait for a minute.. if after that we still are force close the connection and resume this
    // message
    def oldState
    synchronized (lastStateMutex) { oldState = state.lastCommand }
    
    if (logEnable) log.debug "Initial state found: ($oldState)"
    
    if (!((oldState == "Sent Ok") || (oldState == "Send Failed") || (oldState == "Connection Closed") || (oldState == "quit") || (oldState == "Force Closed")))   
      { 
          if (fromRunIn == true)
            {  
                
                def Integer addAmount = Math.floor((Math.random() * 60) + 1)
                def Integer waitTime = 10 + addAmount
                    
                log.debug "2nd attempt to run failed after sleeping... Clearing scheduling, resetting states, re-adding to the queue and aborting!"
                unschedule()
                sendEvent(name: "lastCommand", value: "Force Closed")
                addToQueue(message)
                synchronized (lastStateMutex) { state.lastCommand = "Force Closed" }
                goOn = false
                
                // if debuging on redo the job to turn off
                if (logEnable)
                 {
                  log.info "Debug logging is on. Turning off debug logging in 1/2 hour."
                  runIn(1800,logsOff)
                 }
                     
             runIn(waitTime,"restartFromRunIn", [overwrite: false])   
            }
          
         else
              {
                goOn = false
                  
                 // lgk hubitat slower starting up with connections and multi mails are pilling up.
                 // so to avoid make wait time a random number added to 30 seconds
    
                def Integer addAmount = Math.floor((Math.random() * 60) + 1)
                def Integer waitTime = 30 + addAmount
                  
				if (logEnable || txtEnable) log.info "Existing state ($oldState) indicates last run did not complete. Adding to queue, Waiting $waitTime secs. then trying again!"
                addToQueue(message)
  
                // now reschedule this queue item.
                runIn(waitTime,"restartFromRunIn", [overwrite: false]) 
              }
         }

       if (goOn)
        { 
            if (logEnable) log.debug "Found ok initial state ($oldState) ... going on!"
            
           synchronized (lastStateMutex) { state.lastCommand = "initialConnect"
                                           sendEvent(name: "lastCommand", value: "initialConnect")
                                         }
            
            if (logEnable) synchronized (lastStateMutex) { log.debug "set last command to $state.lastCommand" }
    
	       state.EmailBody = "${message}"
	       state.LastCode = 0
        
           if (logEnable) log.debug "Connecting to ${EmailServer}:${EmailPort}"
	
	       telnetConnect(EmailServer, EmailPort.toInteger(), null, null)  
           // telnetConnect([ terminalType: 'VT100' ], EmailServer, EmailPort.toInteger(), null, null)
        }
}

def sendData(String msg, Integer millsec) {
    
    if (logEnable) log.debug "$msg"
	
	def hubCmd = sendHubCommand(new hubitat.device.HubAction("${msg}", hubitat.device.Protocol.TELNET))
	pauseExecution(millsec)
	
	return hubCmd
}

def parse(String msg) {  
  
    def lastCommand
   
    synchronized (lastStateMutex) { lastCommand = state.lastCommand }
    
    if (logEnable) {
        log.debug "In parse - ${msg}"
        log.debug "lastCommand = $lastCommand"
    }
    
    def first4 = msg.substring(0,4)
    if (first4 == "250-")
    {
        log.debug "Skipping informational command: $msg after ehlo!"
    }
    else
    {
        
    def pair = msg.split(" ")
    def response = pair[0]
    def value = pair[1]
    
   if (logEnable) log.debug "Got server response $response value = $value lastCommand = ($lastCommand)"
    
   if (lastCommand == "initialConnect")
        {
         if (logEnable) log.debug "In initialConnect case"
             if (response == "220")
                 { 
                     sendEvent([name: "telnet", value: "Ok"])
                     if (RequiresEHLO)
                     {
                        if (logEnable) log.debug "Using EHLO instead of HELO!"
                        synchronized (lastStateMutex) { state.lastCommand = "ehlo" }
                        sendEvent(name: "lastCommand", value: "ehlo")      
                        def res1 = sendData("ehlo $state.myHostName",500)
                     }
                     else
                     {
                       synchronized (lastStateMutex) { state.lastCommand = "helo" }
                       sendEvent(name: "lastCommand", value: "helo") 
                       def res1 = sendData("helo $state.myHostName",500)
                     }
                 }
                 else
                 {
                    closeOnError()
                 }
                }
    
   else if (lastCommand == "Auth")
    {
     if (logEnable) log.debug "In auth response looking for 334"
     if (response == "334")
     {
       if (logEnable) log.debug "Got auth response now sending username"
 
       synchronized (lastStateMutex) { state.lastCommand = "Username" }
       sendEvent(name: "lastCommand", value: "Username")
       def res1 = sendData("$Username",500)
     }
    else  
    {
         log.debug "Got bad response for auth = $response"
         synchronized (lastStateMutex) { state.lastCommand = "Send Failed" }
         sendEvent(name: "lastCommand", value: "Send Failed")  
         closeConnection()
     }
    }   

   else if (lastCommand == "Username")
    {
        
     if (logEnable) log.debug "In Username response looking for 334"
        
     if (response == "334")
     {
       if (logEnable) log.debug "Got username response now sending password"
       synchronized (lastStateMutex) { state.lastCommand = "Password" }
       sendEvent(name: "lastCommand", value: "Password")
       def res1 = sendData("$Password",500)
     }
    else  
    {
         log.debug "Got bad response for Username = $response"
         synchronized (lastStateMutex) { state.lastCommand = "Send Failed" }
         sendEvent(name: "lastCommand", value: "Send Failed")  
         closeConnection()
     }
    }   
     
   
    else if ((lastCommand == "helo") || (lastCommand == "ehlo") || (lastCommand == "Password"))
        {
            
        if (logEnable) log.debug "In helo/ehlo/Password case"
            
         if ((response == "250") || (response == "235"))
         {
        
          if (((lastCommand == "helo") || (lastCommand == "ehlo")) && (Authenticate) && (Username) && (Password))
          {
           if (logEnable) log.debug "Trying authentication"
              
           synchronized (lastStateMutex) { state.lastCommand = "Auth" }
           sendEvent(name: "lastCommand", value: "Auth")
           def res1 = sendData("auth login",500)
          }

         else
           {
           if (logEnable) "Either got 250 for helo or 235 for password, Now Sending Message or ehlo"
               
            synchronized (lastStateMutex) { state.lastCommand = "sendmessage" }
            sendEvent(name: "lastCommand", value: "sendmessage") 
                    
            def msgData = "${state.EmailBody}"
	        def emlBody = ""
	        def emlSubject = ""
            def emlDateTime = new Date().format('EEE, dd MMM YYYY H:mm:ss Z',location.timeZone)
               
            def String idP1 = now()
            def String idP2 = java.util.UUID.randomUUID().toString().replaceAll('-', '')
            def String msgId = '<' + idP2 + "." + idP1 + "@2LGKSendmailV2>"
           
          if(msgData.substring(0,1) == "{") {
	             	
		        def slurper = new groovy.json.JsonSlurper()
		        def result = slurper.parseText(msgData)
		        emlBody = result.Body
		        emlSubject = (result.Subject != null ? result.Subject : "")
	        } else {
	           	emlBody = msgData
	        	emlSubject = (Subject != null ? "${Subject}" : "")
	        }   
        
	              def sndMsg =[
                      "MAIL FROM: <${From}>"
                    , "RCPT TO: <${To}>"
	        		, "DATA"
                    , "From: ${From}"
                    , "To: ${To}"
                    , "Date: ${emlDateTime}"
                    , "Message-ID: ${msgId}"
                    , "Subject: ${emlSubject}"  
                    , "MIME-Version: 1.0"
                    , 'Content-Type: text/plain; charset="utf-8"'
                    , "Content-Transfer-Encoding: quoted-printable\r\n"
                    , ""
	        		, "${emlBody}"
            		, ""
	        		, "."
		        	, "quit"
	            ]  
                   
                 def res1 = seqSend(sndMsg,500) 
                 state.messageSent = true  
				 if (debug || txtEnable) log.info "Sent Message: $emlBody"     
         }
         }
         else
         {
            closeOnError()
         }
        }
    
    else if (lastCommand == "sendmessage")
        {
       if (logEnable) log.debug "In send message case"
             if ((response == "220") || (response == "250"))
                 {
					 if (debug || txtEnable) log.info "sending quit"
                     synchronized (lastStateMutex) { state.lastCommand = "quit" } 
                     sendEvent(name: "lastCommand", value: "quit")     
                     def res1 = sendData("quit",500)
                 }
                 else
                 {
                    closeOnError()
                 }
                }
       else if ((lastCommand == "quit") || ((response == "221") && (lastCommand == "other")))
                                            //&& ((value == "bye") || (value == "2.0.0") || (value = "Goodbye"))))
        {
          if (logEnable) log.debug "In quit case"
               if (response == "220" || response == "221" || response == "250")
                 { 
                   if (state.messageSent)
                     {
                      synchronized (lastStateMutex) { state.lastCommand = "Sent Ok" }
                      sendEvent(name: "lastCommand", value: "Sent Ok")  
                     }
                   closeConnection()
                   processQueue()
                 }
                 else
                 {
                      closeOnError()
                 }
           } 
        else 
        {
            if (response == "250" || response == "354")
            //|| response == "221")
            {
               synchronized (lastStateMutex) { state.lastCommand = "other" }
               sendEvent(name: "lastCommand", value: "other")
            }
            else
            {
              closeOnError()
            }
        }         
    }                 
}

def telnetStatus(status) {
    // comment out telent status as it is not working correctly and the system is telling me the socket is closed and other
    // response are still returned on the socket after. ignore stream is closed errors.
    if (status != "receive error: Stream is closed") 
    {    
        if (logEnable) synchronized (lastStateMutex) { log.debug "telnetStatus: ${status} lastcommand is ($state.lastCommand)" }
        sendEvent([name: "telnet", value: "${status}"])
    }
}

def closeConnection()
{
    if (closeTelnet){
                try {
                    telnetClose()
                    synchronized (lastStateMutex) { state.lastCommand = "Connection Closed" }
                    sendEvent(name: "lastCommand", value: "Connection Closed")
                } catch(e) {
                    if (logEnable) log.debug("Connection Closed")
                }
                
			}
}
    
boolean seqSend(msgs, Integer millisec)
{
    if (logEnable) log.debug "in sendData"
  
			msgs.each {
				sendData("${it}",millisec)
			}
			seqSent = true
	return seqSent
}

def closeOnError()
{
     log.debug "Got bad response = $response"
     synchronized (lastStateMutex) { state.lastCommand = "Send Failed" }
     sendEvent(name: "lastCommand", value: "Send Failed")  
     closeConnection()    
}