salesforce-lib/src/main/java/com/streamsets/pipeline/lib/salesforce/BulkRecordCreator.java
private Field pullMap(XMLEventReader reader) throws StageException, XMLStreamException {
    LinkedHashMap<String, Field> map = new LinkedHashMap<>();
    String type = null;
    String fieldValue = null;

    while (reader.hasNext()) {
      XMLEvent event = reader.nextEvent();
      if (event.isStartElement()) {
        if (event.asStartElement().getName().getLocalPart().equals(TYPE)) {
          // Move to content
          event = reader.nextEvent();
          type = event.asCharacters().getData().toLowerCase();
          // Consume closing tag
          reader.nextEvent();
        } else {
          String fieldName = event.asStartElement().getName().getLocalPart();
          Attribute attr = event.asStartElement().getAttributeByName(XSI_TYPE);
          if (attr != null && attr.getValue().equals(S_OBJECT)) {
            // Element is a nested record
            map.put(fieldName, pullMap(reader));
          } else {
            event = reader.nextEvent();
            fieldValue = null;
            switch (event.getEventType()) {
              case XMLEvent.START_ELEMENT:
                // Element is a nested list of records
                // Advance over <done>, <queryLocator> to record list
                while (!(event.isStartElement() && event.asStartElement().getName().getLocalPart().equals(RECORDS))) {
                  event = reader.nextEvent();
                }

                // Read record list
                List<Field> recordList = new ArrayList<>();
                while (event.isStartElement() && event.asStartElement().getName().getLocalPart().equals(RECORDS)) {
                  recordList.add(pullMap(reader));
                  event = reader.nextEvent();
                }
                map.put(fieldName, Field.create(recordList));
                break;
              case XMLEvent.CHARACTERS:
                // Element is a field value
                fieldValue = event.asCharacters().getData();
                // Consume closing tag
                reader.nextEvent();
                // Intentional fall through to next case!
              case XMLEvent.END_ELEMENT:
                // Create the SDC field
                if (type == null) {
                  throw new StageException(Errors.FORCE_38);
                }
                // Is this a relationship to another object?
                com.sforce.soap.partner.Field sfdcField = metadataCache.get(type).getFieldFromRelationship(fieldName);
                if (sfdcField != null) {
                  // See if we already added fields from the related record
                  if (map.get(fieldName) != null) {
                    // We already created this node - don't overwrite it!
                    sfdcField = null;
                  }
                } else {
                  sfdcField = getFieldMetadata(type, fieldName);
                }

                if (sfdcField != null) {
                  Field field = createField(fieldValue, sfdcField);
                  if (conf.createSalesforceNsHeaders) {
                    setHeadersOnField(field, getFieldMetadata(type, fieldName));
                  }

                  map.put(fieldName, field);
                }
                break;
              default:
                throw new StageException(Errors.FORCE_41, event.getEventType());
            }
          }
        }
      } else if (event.isEndElement()) {
        // Done with record
        return Field.createListMap(map);
      }
    }

    throw new StageException(Errors.FORCE_39);
  }
private Field pullMap(XMLEventReader reader) throws StageException, XMLStreamException {
    LinkedHashMap<String, Field> map = new LinkedHashMap<>();
    String type = null;
    String fieldValue = null;

    while (reader.hasNext()) {
      XMLEvent event = reader.nextEvent();
      if (event.isStartElement()) {
        if (event.asStartElement().getName().getLocalPart().equals(TYPE)) {
          // Move to content
          event = reader.nextEvent();
          type = event.asCharacters().getData().toLowerCase();
          // Consume closing tag
          reader.nextEvent();
        } else {
          String fieldName = event.asStartElement().getName().getLocalPart();
          Attribute attr = event.asStartElement().getAttributeByName(XSI_TYPE);
          if (attr != null && attr.getValue().equals(S_OBJECT)) {
            // Element is a nested record
            map.put(fieldName, pullMap(reader));
          } else {
            event = reader.nextEvent();
            fieldValue = null;
            switch (event.getEventType()) {
              case XMLEvent.START_ELEMENT:
                // Element is a nested list of records
                // Advance over <done>, <queryLocator> to record list
                while (!(event.isStartElement() && event.asStartElement().getName().getLocalPart().equals(RECORDS))) {
                  event = reader.nextEvent();
                }

                // Read record list
                List<Field> recordList = new ArrayList<>();
                while (event.isStartElement() && event.asStartElement().getName().getLocalPart().equals(RECORDS)) {
                  recordList.add(pullMap(reader));
                  event = reader.nextEvent();
                }
                map.put(fieldName, Field.create(recordList));
                break;
              case XMLEvent.CHARACTERS:
                // Element is a field value
                fieldValue = event.asCharacters().getData();
                // Consume closing tag
                reader.nextEvent();
                // Intentional fall through to next case!
              case XMLEvent.END_ELEMENT:
                // Create the SDC field
                if (type == null) {
                  throw new StageException(Errors.FORCE_38);
                }
                // Is this a relationship to another object?
                com.sforce.soap.partner.Field sfdcField = metadataCache.get(type).getFieldFromRelationship(fieldName);
                if (sfdcField != null) {
                  // See if we already added fields from the related record
                  if (map.get(fieldName) != null) {
                    // We already created this node - don't overwrite it!
                    sfdcField = null;
                  }
                } else {
                  sfdcField = getFieldMetadata(type, fieldName);
                }

                if (sfdcField != null) {
                  Field field = createField(fieldValue, sfdcField);
                  if (conf.createSalesforceNsHeaders) {
                    setHeadersOnField(field, getFieldMetadata(type, fieldName));
                  }

                  map.put(fieldName, field);
                }
                break;
              default:
                throw new StageException(Errors.FORCE_41, event.getEventType());
            }
          }
        }
      } else if (event.isEndElement()) {
        // Done with record
        return Field.createListMap(map);
      }
    }

    throw new StageException(Errors.FORCE_39);
  }
private Field pullMap ( XMLEventReader reader ) throws StageException , XMLStreamException { LinkedHashMap < String , Field > map = new LinkedHashMap <> ( ) ; String type = null ; String fieldValue = null ; while ( reader . hasNext ( ) ) { XMLEvent event = reader . nextEvent ( ) ; if ( event . isStartElement ( ) ) { if ( event . asStartElement ( ) . getName ( ) . getLocalPart ( ) . equals ( TYPE ) ) { // Move to content event = reader . nextEvent ( ) ; type = event . asCharacters ( ) . getData ( ) . toLowerCase ( ) ; // Consume closing tag reader . nextEvent ( ) ; } else { String fieldName = event . asStartElement ( ) . getName ( ) . getLocalPart ( ) ; Attribute attr = event . asStartElement ( ) . getAttributeByName ( XSI_TYPE ) ; if ( attr != null && attr . getValue ( ) . equals ( S_OBJECT ) ) { // Element is a nested record map . put ( fieldName , pullMap ( reader ) ) ; } else { event = reader . nextEvent ( ) ; fieldValue = null ; switch ( event . getEventType ( ) ) { case XMLEvent . START_ELEMENT : // Element is a nested list of records // Advance over <done>, <queryLocator> to record list while ( ! ( event . isStartElement ( ) && event . asStartElement ( ) . getName ( ) . getLocalPart ( ) . equals ( RECORDS ) ) ) { event = reader . nextEvent ( ) ; } // Read record list List < Field > recordList = new ArrayList <> ( ) ; while ( event . isStartElement ( ) && event . asStartElement ( ) . getName ( ) . getLocalPart ( ) . equals ( RECORDS ) ) { recordList . add ( pullMap ( reader ) ) ; event = reader . nextEvent ( ) ; } map . put ( fieldName , Field . create ( recordList ) ) ; break ; case XMLEvent . CHARACTERS : // Element is a field value fieldValue = event . asCharacters ( ) . getData ( ) ; // Consume closing tag reader . nextEvent ( ) ; // Intentional fall through to next case! case XMLEvent . END_ELEMENT : // Create the SDC field if ( type == null ) { throw new StageException ( Errors . FORCE_38 ) ; } // Is this a relationship to another object? com . sforce . soap . partner . Field sfdcField = metadataCache . get ( type ) . getFieldFromRelationship ( fieldName ) ; if ( sfdcField != null ) { // See if we already added fields from the related record if ( map . get ( fieldName ) != null ) { // We already created this node - don't overwrite it! sfdcField = null ; } } else { sfdcField = getFieldMetadata ( type , fieldName ) ; } if ( sfdcField != null ) { Field field = createField ( fieldValue , sfdcField ) ; if ( conf . createSalesforceNsHeaders ) { setHeadersOnField ( field , getFieldMetadata ( type , fieldName ) ) ; } map . put ( fieldName , field ) ; } break ; default : throw new StageException ( Errors . FORCE_41 , event . getEventType ( ) ) ; } } } } else if ( event . isEndElement ( ) ) { // Done with record return Field . createListMap ( map ) ; } } throw new StageException ( Errors . FORCE_39 ) ; }
When pullMap is called, the caller should have consumed the opening tag for the record
When pullMap is called the caller should have consumed the opening tag for the record