StreamingProxies.cls

public with sharing class StreamingProxies {
    /**
     * @Description This method checks for the existance of proxy records.
     * @Param {List} A list of record Ids.
     * @Return {List} A list of all found StreamingProxy__c records.
     */
    public static List<StreamingProxy__c> recordsExist(List<Id> recordIds) {
        List<StreamingProxy__c> records = [
            SELECT Id, RecordID__c, ObjectType__c FROM StreamingProxy__c WHERE RecordID__c IN :recordIds
        ];
        return records;
    }

    /**
     * @Description This method either inserts or upserts a record for streaming purposes.
     * @Param {SObject} An Salesforce object which we would like to stream or keep track of (ex. via push topics)
     * @Return {StreamingProxy__c} Either the new or update object.
     */
    public static StreamingProxy__c upsertRecord(SObject obj) {
        if(obj == null) return null;

        // sets up the streaming proxy object
        StreamingProxy__c sp = new StreamingProxy__c();
        sp.ObjectType__c = obj.getSObjectType();
        sp.RecordID__c = obj.Id;

        // checks for an existing record
        List<StreamingProxy__c> existing = recordsExist(new List<Id>{ sp.RecordID__c });
        StreamingProxy__c found = existing != null && existing.size() > 0 ? existing[0]: null;

        if(found != null){
            // should preserve the id - so it is treated as an update should it already be there
            sp = found.clone(true, true, true, true);
        }

        // upsert and return
        upsert sp;
        return sp;
    }

    /**
     * @Description This method either inserts or updates a list of records.
     * @Param {List} A list of Salesforce objects that we wish to keep track of (ex. via push topics)
     * @Return {List} A list of StreamingProxy__c objects.
     */
    public static List<StreamingProxy__c> upsertRecords(List<SObject> objs) {
        if(objs == null) return null;

        StreamingProxy__c proxy = null;
        Map<Id, StreamingProxy__c> proxyMap = new Map<Id, StreamingProxy__c>();

        // iterate through the objects to setup their streaming proxy counterparts
        for(SObject obj: objs) {
            proxy = new StreamingProxy__c();
            proxy.ObjectType__c = obj.getSObjectType();
            proxy.RecordID__c = obj.Id;
            proxyMap.put(obj.Id, proxy);
        }

        // apply all existing id's should we find them
        List<StreamingProxy__c> existing = recordsExist(proxyMap.keySet());
        for(StreamingProxy__c sp: existing) {
            if(proxyMap.containsKey(sp.RecordID__c)){
                proxyMap.get(sp.RecordID__c).Id = sp.Id;
            }
        }

        // upsert and return values
        List<StreamingProxy__c> results = proxyMap.values();
        upsert results;

        return results;
    }

    /**
     * @Description This method deletes a streaming record should it exist.
     * @Param {SObject} The SObject that may have a paired streaming object.
     * @Return {StreamingProxy__c} The deleted object entry.
     */
    public static StreamingProxy__c deleteRecord(SObject obj) {
        if(obj == null) return null;

        // sets up the streaming proxy object
        StreamingProxy__c sp = new StreamingProxy__c();
        sp.RecordID__c = obj.Id;

        // checks for an existing record
        List<StreamingProxy__c> existing = recordsExist(new List<Id>{ sp.RecordID__c });

        if(!existing.isEmpty()){
            delete existing;
            return existing[0];
        }
        return null;
    }

    /**
     * @Description This method deletes a list of records should they exist.
     * @Param {List} A list of Salesforce objects that we desire to delete.
     * @Return {List} The list of deleted objects.
     */
    public static List<StreamingProxy__c> deleteRecords(List<SObject> objs) {
        if(objs == null) return null;

        // gather the record ids
        List<Id> recordIds = new List<Id>();
        for(SObject obj: objs) {
            recordIds.add(obj.Id);
        }

        // checks for existing records
        List<StreamingProxy__c> existing = recordsExist(recordIds);
        if(!existing.isEmpty()) {
            delete existing;
        }
        return existing;
    }
}